Verify production CTAS with the default write setting

ydb-trino-adapter/README.md

@@ -60,9 +60,17 @@ AS SELECT tenant, event_id, payload FROM events;
 ```
 
 Имена ключей в CTAS относятся к выходным столбцам запроса. Отсутствующий,
-пустой, повторяющийся или неизвестный `primary_key` отклоняется. Текущая
-integration-проверка CTAS использует `insert.non-transactional-insert.enabled=true`;
-transactional staging этой проверкой не подтверждается.
+пустой, повторяющийся или неизвестный `primary_key` отклоняется. CTAS адаптера
+работает при обоих значениях `insert.non-transactional-insert.enabled`.
+
+При значении `false` (по умолчанию в Trino 483) обычный `INSERT` не работает:
+Trino пытается создать промежуточную таблицу запросом `CREATE TABLE tmp AS SELECT
+... WHERE 0 = 1`, который YDB отклоняет на `AS`. Эта форма не соответствует
+[синтаксису YQL CTAS](https://ydb.tech/docs/en/yql/reference/syntax/create_table/as_select):
+он требует первичный ключ и поддерживается для колоночных таблиц. Для `INSERT`
+задайте `insert.non-transactional-insert.enabled=true` в свойствах каталога.
+Тогда Trino пишет непосредственно в целевую таблицу; при сбое запись может
+остаться частичной.
 
 ## Текст и байты
 

ydb-trino-adapter/src/test/java/tech/ydb/trino/TestYdbDefaultWriteMode.java

@@ -0,0 +1,45 @@
+package tech.ydb.trino;
+
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import tech.ydb.test.junit5.YdbHelperExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class TestYdbDefaultWriteMode extends AbstractTestQueryFramework {
+    @RegisterExtension
+    static final YdbHelperExtension ydb = new YdbHelperExtension();
+
+    @Override
+    protected QueryRunner createQueryRunner() throws Exception {
+        return YdbQueryRunner.builder(ydb)
+                .addConnectorProperty("insert.non-transactional-insert.enabled", "false")
+                .useProductionClient()
+                .build();
+    }
+
+    @Test
+    public void testCreateTableAsSelect() {
+        String tableName = "default_mode_ctas";
+        String duplicateTableName = "default_mode_ctas_duplicate";
+        try {
+            assertUpdate("CREATE TABLE " + tableName + " (id, payload) WITH (primary_key = ARRAY['id']) " +
+                    "AS VALUES (BIGINT '1', 'a'), (BIGINT '2', 'b')", 2);
+            assertQuery("SELECT * FROM " + tableName,
+                    "VALUES (CAST(1 AS BIGINT), 'a'), (CAST(2 AS BIGINT), 'b')");
+
+            var tablesBefore = computeActual("SHOW TABLES").getOnlyColumnAsSet();
+            assertThat(query("CREATE TABLE " + duplicateTableName + " (id, payload) " +
+                    "WITH (primary_key = ARRAY['id']) AS VALUES (BIGINT '1', 'a'), (BIGINT '1', 'b')"))
+                    .failure();
+            assertThat(computeActual("SHOW TABLES").getOnlyColumnAsSet()).isEqualTo(tablesBefore);
+            assertThat(getQueryRunner().tableExists(getSession(), duplicateTableName)).isFalse();
+        }
+        finally {
+            assertUpdate("DROP TABLE IF EXISTS " + tableName);
+            assertUpdate("DROP TABLE IF EXISTS " + duplicateTableName);
+        }
+    }
+}