[ https://issues.apache.org/jira/browse/HIVE-25034?focusedWorklogId=598030&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-598030 ]
ASF GitHub Bot logged work on HIVE-25034: ----------------------------------------- Author: ASF GitHub Bot Created on: 17/May/21 18:25 Start Date: 17/May/21 18:25 Worklog Time Spent: 10m Work Description: pvary commented on a change in pull request #2243: URL: https://github.com/apache/hive/pull/2243#discussion_r633483880 ########## File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java ########## @@ -138,6 +148,17 @@ public void initialize(@Nullable Configuration configuration, Properties serDePr } } + private void createTableForCTAS(Configuration configuration, Properties serDeProperties) { + serDeProperties.setProperty(TableProperties.ENGINE_HIVE_ENABLED, "true"); + serDeProperties.setProperty(InputFormatConfig.TABLE_SCHEMA, SchemaParser.toJson(tableSchema)); + Catalogs.createTable(configuration, serDeProperties); + // set these in the global conf so that we can rollback the table in the lifecycle hook in case of failures Review comment: A good candidate to put something into `QueryInfo`, or somewhere else ########## File path: iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerWithEngine.java ########## @@ -514,6 +519,75 @@ public void testInsertOverwritePartitionedTable() throws IOException { HiveIcebergTestUtils.validateData(table, expected, 0); } + @Test + public void testCTASFromHiveTable() { + Assume.assumeTrue("CTAS target table is supported only for HiveCatalog tables", + testTableType == TestTables.TestTableType.HIVE_CATALOG); Review comment: Why? What is blocking us? ########## File path: iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerWithEngine.java ########## @@ -514,6 +519,75 @@ public void testInsertOverwritePartitionedTable() throws IOException { HiveIcebergTestUtils.validateData(table, expected, 0); } + @Test + public void testCTASFromHiveTable() { + Assume.assumeTrue("CTAS target table is supported only for HiveCatalog tables", + testTableType == TestTables.TestTableType.HIVE_CATALOG); + + shell.executeStatement("CREATE TABLE source (id bigint, name string) PARTITIONED BY (dept string) STORED AS ORC"); + shell.executeStatement("INSERT INTO source VALUES (1, 'Mike', 'HR'), (2, 'Linda', 'Finance')"); + + shell.executeStatement(String.format( + "CREATE TABLE target STORED BY '%s' %s TBLPROPERTIES ('%s'='%s') AS SELECT * FROM source", + HiveIcebergStorageHandler.class.getName(), + testTables.locationForCreateTableSQL(TableIdentifier.of("default", "target")), + TableProperties.DEFAULT_FILE_FORMAT, fileFormat)); + + List<Object[]> objects = shell.executeStatement("SELECT * FROM target ORDER BY id"); + Assert.assertEquals(2, objects.size()); + Assert.assertArrayEquals(new Object[]{1L, "Mike", "HR"}, objects.get(0)); + Assert.assertArrayEquals(new Object[]{2L, "Linda", "Finance"}, objects.get(1)); + } + + @Test + public void testCTASFromDifferentIcebergCatalog() { Review comment: Would it be better placed here `TestHiveIcebergStorageHandlerWithMultipleCatalogs`? ########## File path: iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveShell.java ########## @@ -216,6 +216,9 @@ private HiveConf initializeConf() { // enables vectorization on Tez hiveConf.set("tez.mrreader.config.update.properties", "hive.io.file.readcolumn.names,hive.io.file.readcolumn.ids"); + // set lifecycle hooks + hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_LIFETIME_HOOKS, HiveIcebergCTASHook.class.getName()); Review comment: I have seen several hooks already. Maybe we would like to keep them as a single class? ########## File path: iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveShell.java ########## @@ -216,6 +216,9 @@ private HiveConf initializeConf() { // enables vectorization on Tez hiveConf.set("tez.mrreader.config.update.properties", "hive.io.file.readcolumn.names,hive.io.file.readcolumn.ids"); + // set lifecycle hooks + hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_LIFETIME_HOOKS, HiveIcebergCTASHook.class.getName()); Review comment: I have seen another Iceberg hook already. Maybe we would like to keep them as a single class? Maybe even if they are implementing different interfaces? - Just an idea which I am playing around ########## File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java ########## @@ -361,7 +361,7 @@ private void collectCommitInformation(TezWork work) throws IOException, TezExcep .filter(name -> name.endsWith("HiveIcebergNoJobCommitter")).isPresent(); // we should only consider jobs with Iceberg output committer and a data sink if (hasIcebergCommitter && !vertex.getDataSinks().isEmpty()) { - String tableLocationRoot = jobConf.get("location"); + String tableLocationRoot = jobConf.get("iceberg.mr.table.location"); Review comment: What is this change? Shall we use a constant? ########## File path: ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java ########## @@ -7759,6 +7759,18 @@ protected Operator genFileSinkPlan(String dest, QB qb, Operator input) .getMsg(destinationPath.toUri().toString())); } } + // handle direct insert CTAS case + // for direct insert CTAS, the table creation DDL is not added to the task plan in TaskCompiler, + // therefore we need to add the InsertHook here manually so that HiveMetaHook#commitInsertTable is called Review comment: Where do we add the InsertHook here? I do not really grok the comment. ########## File path: ql/src/java/org/apache/hadoop/hive/ql/parse/TaskCompiler.java ########## @@ -134,6 +135,16 @@ public void compile(final ParseContext pCtx, boolean isCStats = pCtx.getQueryProperties().isAnalyzeRewrite(); int outerQueryLimit = pCtx.getQueryProperties().getOuterQueryLimit(); + boolean directInsertCtas = false; + if (pCtx.getCreateTable() != null && pCtx.getCreateTable().getStorageHandler() != null) { + try { + directInsertCtas = + HiveUtils.getStorageHandler(conf, pCtx.getCreateTable().getStorageHandler()).directInsertCTAS(); + } catch (HiveException e) { Review comment: When do we get `HiveException`? ########## File path: ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java ########## @@ -7759,6 +7759,18 @@ protected Operator genFileSinkPlan(String dest, QB qb, Operator input) .getMsg(destinationPath.toUri().toString())); } } + // handle direct insert CTAS case + // for direct insert CTAS, the table creation DDL is not added to the task plan in TaskCompiler, + // therefore we need to add the InsertHook here manually so that HiveMetaHook#commitInsertTable is called + if (qb.isCTAS() && tableDesc != null && tableDesc.getStorageHandler() != null) { + try { + if (HiveUtils.getStorageHandler(conf, tableDesc.getStorageHandler()).directInsertCTAS()) { + createPreInsertDesc(destinationTable, false); + } + } catch (HiveException e) { Review comment: When do we get `HiveException`? Why swallow it? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking ------------------- Worklog Id: (was: 598030) Time Spent: 1h 40m (was: 1.5h) > Implement CTAS for Iceberg > -------------------------- > > Key: HIVE-25034 > URL: https://issues.apache.org/jira/browse/HIVE-25034 > Project: Hive > Issue Type: New Feature > Reporter: Marton Bod > Assignee: Marton Bod > Priority: Major > Labels: pull-request-available > Time Spent: 1h 40m > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.3.4#803005)