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..f4c738127733 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergMerge.java @@ -0,0 +1,92 @@ +/* + * 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.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(); + + + @Override + protected void validateTestParams() { + } + + @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());