diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergMerge.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergMerge.java new file mode 100644 index 000000000000..d7f00ab6a167 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergMerge.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive; + +import java.util.List; +import org.apache.hadoop.hive.ql.parse.SemanticException; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.mr.TestHelper; +import org.apache.iceberg.mr.hive.test.utils.HiveIcebergTestUtils; +import org.apache.iceberg.relocated.com.google.common.base.Throwables; +import org.apache.iceberg.types.Types; +import org.junit.Assert; +import org.junit.Test; + +import static org.apache.iceberg.types.Types.NestedField.optional; + +/** + * Tests Merge Related SQL features. + */ +public class TestHiveIcebergMerge extends HiveIcebergStorageHandlerWithEngineBase { + + static final Schema SCHEMA = new Schema( + optional(1, "a", Types.IntegerType.get()), + optional(2, "b", Types.IntegerType.get(), "This is first name") + ); + + static final List TGT_RECORDS = TestHelper.RecordsBuilder.newInstance(SCHEMA) + .add(0, 1) + .add(9, 9) + .build(); + + @Test + public void testMergeIntoOnClauseColumnsNoAssignedTables() { + testTables.createTable(shell, "tgt", SCHEMA, + PartitionSpec.unpartitioned(), fileFormat, TGT_RECORDS, formatVersion); + shell.executeStatement("alter table default.tgt set tblproperties('write.merge.mode'='copy-on-write')"); + shell.executeStatement("drop table if exists default.src"); + shell.executeStatement("create table default.src (a int, b int) stored as orc"); + shell.executeStatement("use default"); + String sql = "MERGE INTO tgt using src on a= src.a when matched then update set b=10"; + try { + shell.executeStatement(sql); + Assert.assertFalse(true); // place we should not touch + } catch (Throwable ex) { + Throwable cause = Throwables.getRootCause(ex); + Assert.assertTrue(cause instanceof SemanticException); + } + } + + @Test + public void testMergeIntoOnClauseColumns() { + testTables.createTable(shell, "tgt", SCHEMA, + PartitionSpec.unpartitioned(), fileFormat, TGT_RECORDS, formatVersion); + shell.executeStatement("alter table default.tgt set tblproperties('write.merge.mode'='copy-on-write')"); + shell.executeStatement("drop table if exists default.src"); + shell.executeStatement("create table default.src (a int, b int) stored as orc"); + shell.executeStatement("insert into table default.src select 0, 10"); + shell.executeStatement("use default"); + shell.executeStatement("merge into tgt using src on tgt.a = src.a when matched then update set b=src.b"); + List objects = shell.executeStatement("SELECT * FROM tgt ORDER BY a"); + Assert.assertEquals(2, objects.size()); + List expected = TestHelper.RecordsBuilder.newInstance(SCHEMA) + .add(0, 10) + .add(9, 9) + .build(); + HiveIcebergTestUtils.validateData(expected, + HiveIcebergTestUtils.valueForRow(SCHEMA, objects), 0); + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/MergeSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/MergeSemanticAnalyzer.java index 4987384fbc02..020cdb79b6bd 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/MergeSemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/MergeSemanticAnalyzer.java @@ -130,7 +130,15 @@ WHEN NOT MATCHED THEN INSERT VALUES (source.a2, source.b2) .sourceName(sourceName) .sourceAlias(getSourceAlias(source, sourceName)) .onClauseAsText(onClauseAsText); - + + OnClauseAnalyzer oca = new OnClauseAnalyzer(onClause, targetTable, targetAlias, + conf, onClauseAsText); + oca.analyze(); + // unresolved columns are not allowed in the on clause to avoid wrong results + if (!oca.unresolvedColumns.isEmpty()) { + throw new SemanticException("UnResolvedColumns exist: " + String.join(",", oca.unresolvedColumns) + + ". We should assign a table name to each column in the ON clause like tbl.col."); + } int whenClauseBegins = 3; boolean hasHint = false; // query hint @@ -159,12 +167,7 @@ WHEN NOT MATCHED THEN INSERT VALUES (source.a2, source.b2) for (ASTNode whenClause : whenClauses) { switch (getWhenClauseOperation(whenClause).getType()) { case HiveParser.TOK_INSERT: - numInsertClauses++; - - OnClauseAnalyzer oca = new OnClauseAnalyzer(onClause, targetTable, targetAlias, - conf, onClauseAsText); - oca.analyze(); - + numInsertClauses++; mergeStatementBuilder.addWhenClause( handleInsert(whenClause, oca.getPredicate(), targetTable)) .onClausePredicate(oca.getPredicate()); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java index f547b9366632..e27d7f38c0c8 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java @@ -1114,19 +1114,19 @@ public void testQuotedIdentifier() throws Exception { "`d?*de e` decimal(5,2)," + "vc varchar(128)) clustered by (i) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')"); runStatementOnDriver("create table " + src + "(gh int, j decimal(5,2), k varchar(128))"); - runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on i=gh " + + runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on `d/8`.i=`a/b`.gh " + "\nwhen matched and i > 5 then delete " + "\nwhen matched then update set vc='blah' " + "\nwhen not matched then insert values(1,2.1,'baz')"); - runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on i=gh " + + runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on `d/8`.i=`a/b`.gh " + "\nwhen matched and i > 5 then delete " + "\nwhen matched then update set vc='blah', `d?*de e` = current_timestamp() " + "\nwhen not matched then insert values(1,2.1, concat('baz', current_timestamp()))"); - runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on i=gh " + + runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on `d/8`.i=`a/b`.gh " + "\nwhen matched and i > 5 then delete " + "\nwhen matched then update set vc='blah' " + "\nwhen not matched then insert values(1,2.1,'a\\b')"); - runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on i=gh " + + runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on `d/8`.i=`a/b`.gh " + "\nwhen matched and i > 5 then delete " + "\nwhen matched then update set vc='∆∋'" + "\nwhen not matched then insert values(`a/b`.gh,`a/b`.j,'c\\t')"); @@ -1141,11 +1141,11 @@ public void testQuotedIdentifier2() throws Exception { "`d?*de e` decimal(5,2)," + "vc varchar(128)) clustered by (i) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')"); runStatementOnDriver("create table " + src + "(`g/h` int, j decimal(5,2), k varchar(128))"); - runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on i=`g/h`" + + runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on `d/8`.i=`a/b`.`g/h`" + "\nwhen matched and `g/h` > 5 then delete " + "\nwhen matched and `g/h` < 0 then update set vc='∆∋', `d?*de e` = `d?*de e` * j + 1" + "\nwhen not matched and `d?*de e` <> 0 then insert values(`a/b`.`g/h`,`a/b`.j,`a/b`.k)"); - runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on i=`g/h`" + + runStatementOnDriver("merge into " + target + " as `d/8` using " + src + " as `a/b` on `d/8`.i=`a/b`.`g/h`" + "\nwhen matched and `g/h` > 5 then delete" + "\n when matched and `g/h` < 0 then update set vc='∆∋' , `d?*de e` = `d?*de e` * j + 1 " + "\n when not matched and `d?*de e` <> 0 then insert values(`a/b`.`g/h`,`a/b`.j,`a/b`.k)"); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java index cd135d93130d..2d2cd314f638 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java @@ -1788,7 +1788,8 @@ public void testMerge() throws Exception { List r = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a,b"); Assert.assertEquals(stringifyValues(vals), r); String query = "merge into " + Table.ACIDTBL + - " using " + Table.NONACIDPART2 + " source ON " + Table.ACIDTBL + ".a = a2 and b + 1 = source.b2 + 1 " + + " using " + Table.NONACIDPART2 + " source ON " + Table.ACIDTBL + ".a = source.a2 and " + + Table.ACIDTBL +".b + 1 = source.b2 + 1 " + "WHEN MATCHED THEN UPDATE set b = source.b2 " + "WHEN NOT MATCHED THEN INSERT VALUES(source.a2, source.b2)"; runStatementOnDriver(query);