This is an automated email from the ASF dual-hosted git repository.
ayushtkn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 53f8436b5e5 HIVE-29754: Remove empty HIVE_UNION_SUBDIR moved
directories after INSERT OVERWRITE UNION ALL (#6625)
53f8436b5e5 is described below
commit 53f8436b5e576394e3ac3b412eba833edc20e2f3
Author: Indhumathi <[email protected]>
AuthorDate: Tue Jul 28 20:26:55 2026 +0530
HIVE-29754: Remove empty HIVE_UNION_SUBDIR moved directories after INSERT
OVERWRITE UNION ALL (#6625)
---
.../test/resources/testconfiguration.properties | 1 +
.../hive/ql/exec/AbstractFileMergeOperator.java | 2 +-
.../hadoop/hive/ql/exec/FileSinkOperator.java | 2 +-
.../org/apache/hadoop/hive/ql/exec/Utilities.java | 174 +++++++-------
.../io/rcfile/truncate/ColumnTruncateMapper.java | 3 +-
.../union_all_empty_branch_no_tmp_dir.q | 87 +++++++
.../tez/union_all_empty_branch_no_tmp_dir.q.out | 251 +++++++++++++++++++++
7 files changed, 434 insertions(+), 86 deletions(-)
diff --git a/itests/src/test/resources/testconfiguration.properties
b/itests/src/test/resources/testconfiguration.properties
index 8bfda2ea2af..efdbebcf9b5 100644
--- a/itests/src/test/resources/testconfiguration.properties
+++ b/itests/src/test/resources/testconfiguration.properties
@@ -39,6 +39,7 @@ minitez.query.files=\
tez_union_udtf.q,\
tez_union_with_udf.q,\
udf_configurable.q,\
+ union_all_empty_branch_no_tmp_dir.q,\
update_orig_table.q,\
vector_join_part_col_char.q,\
vector_non_string_partition.q
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
index 454fc3d5c97..e833e4f4167 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
@@ -324,7 +324,7 @@ public void jobCloseOp(Configuration hconf, boolean success)
if (!isMmTable) {
Path backupPath = backupOutputPath(fs, outputDir);
Utilities.mvFileToFinalPath(
- outputDir, null, hconf, success, LOG, conf.getDpCtx(), null,
reporter);
+ outputDir, null, hconf, success, conf.getDpCtx(), null, reporter);
if (success) {
LOG.info("jobCloseOp moved merged files to output dir: " +
outputDir);
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
index a3d48f36a31..440a263c6d3 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
@@ -1620,7 +1620,7 @@ public void jobCloseOp(Configuration hconf, boolean
success)
Utilities.FILE_OP_LOGGER.trace("jobCloseOp using specPath " +
specPath);
}
if (!conf.isMmTable() && !conf.isDirectInsert()) {
- Utilities.mvFileToFinalPath(specPath, unionSuffix, hconf, success,
LOG, dpCtx, conf, reporter);
+ Utilities.mvFileToFinalPath(specPath, unionSuffix, hconf, success,
dpCtx, conf, reporter);
} else {
int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
lbLevels = lbCtx == null ? 0 :
lbCtx.calculateListBucketingLevel();
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
index 8a72a22b97f..125bc55ea68 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
@@ -1434,21 +1434,16 @@ public static FileStatus[] listStatusIfExists(Path
path, FileSystem fs) throws I
}
public static void mvFileToFinalPath(Path specPath, String unionSuffix,
Configuration hconf,
- boolean success, Logger log,
DynamicPartitionCtx dpCtx, FileSinkDesc conf,
+ boolean success, DynamicPartitionCtx
dpCtx, FileSinkDesc conf,
Reporter reporter) throws IOException,
HiveException {
- // There are following two paths this could could take based on the value
of shouldAvoidRename
- // shouldAvoidRename indicate if tmpPath should be renamed/moved or now.
- // if false:
- // Skip renaming/moving the tmpPath
- // Deduplicate and keep a list of files
- // Pass on the list of files to conf (to be used later by fetch
operator)
- // if true:
- // 1) Rename tmpPath to a new directory name to prevent additional
files
- // from being added by runaway processes.
- // 2) Remove duplicates from the temp directory
- // 3) Rename/move the temp directory to specPath
+ // On success, behavior is controlled by whether the staging directory
should be renamed (see computeAvoidRename).
+ // For SELECT statements and CTAS / Create MV on blob storage, staging
directory is not renamed;
+ // files are deduplicated and either passed to the fetch operator (SELECT)
or moved in place (CTAS) via
+ // moveKeptFiles. For INSERT, LOAD and similar write statements:
renameTmpIfNeeded fences runaway processes
+ // by renaming tmpPath to *.moved, then removeDuplicatesAndFillBuckets +
moveKeptFiles move files to specPath.
+ // If the branch produced no output, then .moved staging dir is deleted.
FileSystem fs = specPath.getFileSystem(hconf);
Path tmpPath = Utilities.toTempPath(specPath);
@@ -1456,78 +1451,21 @@ public static void mvFileToFinalPath(Path specPath,
String unionSuffix, Configur
if (!StringUtils.isEmpty(unionSuffix)) {
specPath = specPath.getParent();
}
- PerfLogger perfLogger = SessionState.getPerfLogger();
- boolean isBlobStorage = BlobStorageUtils.isBlobStorageFileSystem(hconf,
fs);
- boolean avoidRename = false;
- boolean shouldAvoidRename = shouldAvoidRename(conf, hconf);
-
- if(isBlobStorage && (shouldAvoidRename|| ((conf != null) &&
conf.isCTASorCM()))
- || (!isBlobStorage && shouldAvoidRename)) {
- avoidRename = true;
- }
+ boolean avoidRename = computeAvoidRename(conf, hconf, fs);
if (success) {
- if (!avoidRename && fs.exists(tmpPath)) {
- // 1) Rename tmpPath to a new directory name to prevent additional
files
- // from being added by runaway processes.
- // this is only done for all statements except SELECT, CTAS and Create
MV
- Path tmpPathOriginal = tmpPath;
- tmpPath = new Path(tmpPath.getParent(), tmpPath.getName() + ".moved");
- LOG.debug("shouldAvoidRename is false therefore moving/renaming " +
tmpPathOriginal + " to " + tmpPath);
- perfLogger.perfLogBegin("FileSinkOperator", "rename");
- Utilities.rename(fs, tmpPathOriginal, tmpPath);
- perfLogger.perfLogEnd("FileSinkOperator", "rename");
- }
-
- // Remove duplicates from tmpPath
+ tmpPath = renameTmpIfNeeded(fs, tmpPath, avoidRename);
List<FileStatus> statusList = HiveStatsUtils.getFileStatusRecurse(
- tmpPath, ((dpCtx == null) ? 1 : dpCtx.getNumDPCols()), fs);
- FileStatus[] statuses = statusList.toArray(new
FileStatus[statusList.size()]);
- if(statuses != null && statuses.length > 0) {
- Set<FileStatus> filesKept = new HashSet<>();
- perfLogger.perfLogBegin("FileSinkOperator",
"RemoveTempOrDuplicateFiles");
- // remove any tmp file or double-committed output files
- int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
- numBuckets = (conf != null && conf.getTable() != null) ?
conf.getTable().getNumBuckets() : 0;
- List<Path> emptyBuckets = removeTempOrDuplicateFiles(
- fs, statuses, unionSuffix, dpLevels, numBuckets, hconf, null, 0,
false, filesKept, false);
- perfLogger.perfLogEnd("FileSinkOperator",
"RemoveTempOrDuplicateFiles");
- // create empty buckets if necessary
- if (!emptyBuckets.isEmpty()) {
- perfLogger.perfLogBegin("FileSinkOperator", "CreateEmptyBuckets");
- createEmptyBuckets(
- hconf, emptyBuckets, conf.getCompressed(), conf.getTableInfo(),
reporter);
- for(Path p:emptyBuckets) {
- FileStatus[] items = fs.listStatus(p);
- filesKept.addAll(Arrays.asList(items));
- }
- perfLogger.perfLogEnd("FileSinkOperator", "CreateEmptyBuckets");
- }
-
- // move to the file destination
- Utilities.FILE_OP_LOGGER.trace("Moving tmp dir: {} to: {}", tmpPath,
specPath);
- if(shouldAvoidRename(conf, hconf)){
- // for SELECT statements
- LOG.debug("Skipping rename/move files. Files to be kept are: " +
filesKept.toString());
- conf.getFilesToFetch().addAll(filesKept);
- } else if (conf !=null && conf.isCTASorCM() && isBlobStorage) {
- // for CTAS or Create MV statements
- perfLogger.perfLogBegin("FileSinkOperator",
"moveSpecifiedFileStatus");
- LOG.debug("CTAS/Create MV: Files being renamed: " +
filesKept.toString());
- if (conf.getTable() != null &&
conf.getTable().getTableType().equals(TableType.EXTERNAL_TABLE)) {
- // Do this optimisation only for External tables.
- createFileList(filesKept, tmpPath, specPath, fs);
- } else {
- Set<String> filesKeptPaths = filesKept.stream().map(x ->
x.getPath().toString()).collect(Collectors.toSet());
- moveSpecifiedFilesInParallel(hconf, fs, tmpPath, specPath,
filesKeptPaths);
- }
- perfLogger.perfLogEnd("FileSinkOperator", "moveSpecifiedFileStatus");
- } else {
- // for rest of the statement e.g. INSERT, LOAD etc
- perfLogger.perfLogBegin("FileSinkOperator", "RenameOrMoveFiles");
- LOG.debug("Final renaming/moving. Source: " + tmpPath + "
.Destination: " + specPath);
- renameOrMoveFilesInParallel(hconf, fs, tmpPath, specPath);
- perfLogger.perfLogEnd("FileSinkOperator", "RenameOrMoveFiles");
- }
+ tmpPath, (dpCtx == null ? 1 : dpCtx.getNumDPCols()), fs);
+ FileStatus[] statuses = statusList.toArray(new FileStatus[0]);
+ if (statuses.length > 0) {
+ Set<FileStatus> filesKept = removeDuplicatesAndFillBuckets(
+ hconf, fs, statuses, unionSuffix, dpCtx, conf, reporter);
+ moveKeptFiles(hconf, fs, tmpPath, specPath, conf, filesKept);
+ } else if (!avoidRename) {
+ // Empty UNION branch: .moved staging dir exists but contains no
output files - delete it
+ // (e.g. UNION ALL branch produced 0 rows)
+ Utilities.FILE_OP_LOGGER.debug("Deleting empty staging directory after
commit: {}", tmpPath);
+ fs.delete(tmpPath, true);
}
} else {
Utilities.FILE_OP_LOGGER.trace("deleting tmpPath {}", tmpPath);
@@ -1537,6 +1475,78 @@ public static void mvFileToFinalPath(Path specPath,
String unionSuffix, Configur
fs.delete(taskTmpPath, true);
}
+ private static boolean computeAvoidRename(FileSinkDesc conf, Configuration
hconf, FileSystem fs) {
+ boolean isBlobStorage = BlobStorageUtils.isBlobStorageFileSystem(hconf,
fs);
+ boolean shouldAvoidRename = shouldAvoidRename(conf, hconf);
+ return (isBlobStorage && (shouldAvoidRename || (conf != null &&
conf.isCTASorCM())))
+ || (!isBlobStorage && shouldAvoidRename);
+ }
+
+ private static Path renameTmpIfNeeded(FileSystem fs, Path tmpPath, boolean
avoidRename)
+ throws IOException, HiveException {
+ if (!avoidRename && fs.exists(tmpPath)) {
+ Path moved = new Path(tmpPath.getParent(), tmpPath.getName() + ".moved");
+ LOG.debug("shouldAvoidRename is false therefore moving/renaming {} to
{}", tmpPath, moved);
+ PerfLogger perfLogger = SessionState.getPerfLogger();
+ perfLogger.perfLogBegin("FileSinkOperator", "rename");
+ Utilities.rename(fs, tmpPath, moved);
+ perfLogger.perfLogEnd("FileSinkOperator", "rename");
+ return moved;
+ }
+ return tmpPath;
+ }
+
+ private static Set<FileStatus> removeDuplicatesAndFillBuckets(Configuration
hconf, FileSystem fs,
+ FileStatus[] statuses, String unionSuffix, DynamicPartitionCtx dpCtx,
+ FileSinkDesc conf, Reporter reporter) throws IOException, HiveException {
+ Set<FileStatus> filesKept = new HashSet<>();
+ PerfLogger perfLogger = SessionState.getPerfLogger();
+ int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols();
+ int numBuckets = (conf != null && conf.getTable() != null) ?
conf.getTable().getNumBuckets() : 0;
+ perfLogger.perfLogBegin("FileSinkOperator", "RemoveTempOrDuplicateFiles");
+ List<Path> emptyBuckets = removeTempOrDuplicateFiles(
+ fs, statuses, unionSuffix, dpLevels, numBuckets, hconf, null, 0,
false, filesKept, false);
+ perfLogger.perfLogEnd("FileSinkOperator", "RemoveTempOrDuplicateFiles");
+ if (conf != null && !emptyBuckets.isEmpty()) {
+ perfLogger.perfLogBegin("FileSinkOperator", "CreateEmptyBuckets");
+ createEmptyBuckets(hconf, emptyBuckets, conf.getCompressed(),
conf.getTableInfo(), reporter);
+ for (Path p : emptyBuckets) {
+ filesKept.addAll(Arrays.asList(fs.listStatus(p)));
+ }
+ perfLogger.perfLogEnd("FileSinkOperator", "CreateEmptyBuckets");
+ }
+ return filesKept;
+ }
+
+ private static void moveKeptFiles(Configuration hconf, FileSystem fs, Path
tmpPath,
+ Path specPath, FileSinkDesc conf, Set<FileStatus> filesKept) throws
IOException, HiveException {
+ Utilities.FILE_OP_LOGGER.trace("Moving tmp dir: {} to: {}", tmpPath,
specPath);
+ PerfLogger perfLogger = SessionState.getPerfLogger();
+ if (shouldAvoidRename(conf, hconf)) {
+ // SELECT statements: pass files to fetch operator
+ LOG.debug("Skipping rename/move files. Files to be kept are: {}",
filesKept);
+ conf.getFilesToFetch().addAll(filesKept);
+ } else if (conf != null && conf.isCTASorCM()
+ && BlobStorageUtils.isBlobStorageFileSystem(hconf, fs)) {
+ // CTAS / Create MV on blob storage
+ perfLogger.perfLogBegin("FileSinkOperator", "moveSpecifiedFileStatus");
+ LOG.debug("CTAS/Create MV: Files being renamed: {}", filesKept);
+ if (conf.getTable() != null &&
conf.getTable().getTableType().equals(TableType.EXTERNAL_TABLE)) {
+ createFileList(filesKept, tmpPath, specPath, fs);
+ } else {
+ Set<String> paths = filesKept.stream().map(x ->
x.getPath().toString()).collect(Collectors.toSet());
+ moveSpecifiedFilesInParallel(hconf, fs, tmpPath, specPath, paths);
+ }
+ perfLogger.perfLogEnd("FileSinkOperator", "moveSpecifiedFileStatus");
+ } else {
+ // INSERT, LOAD, etc.
+ perfLogger.perfLogBegin("FileSinkOperator", "RenameOrMoveFiles");
+ LOG.debug("Final renaming/moving. Source: {} .Destination: {}", tmpPath,
specPath);
+ renameOrMoveFilesInParallel(hconf, fs, tmpPath, specPath);
+ perfLogger.perfLogEnd("FileSinkOperator", "RenameOrMoveFiles");
+ }
+ }
+
private static void createFileList(Set<FileStatus> filesKept, Path srcPath,
Path targetPath, FileSystem fs)
throws IOException {
try (FSDataOutputStream outStream = fs.create(new Path(targetPath,
BLOB_MANIFEST_FILE))) {
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
b/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
index 736b9e39cb2..963195e235d 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
@@ -234,8 +234,7 @@ public static void jobClose(Path outputPath, boolean
success, JobConf job,
) throws HiveException, IOException {
FileSystem fs = outputPath.getFileSystem(job);
Path backupPath = backupOutputPath(fs, outputPath, job);
- Utilities.mvFileToFinalPath(outputPath, null, job, success, LOG,
dynPartCtx, null,
- reporter);
+ Utilities.mvFileToFinalPath(outputPath, null, job, success, dynPartCtx,
null, reporter);
if (backupPath != null) {
fs.delete(backupPath, true);
}
diff --git
a/ql/src/test/queries/clientpositive/union_all_empty_branch_no_tmp_dir.q
b/ql/src/test/queries/clientpositive/union_all_empty_branch_no_tmp_dir.q
new file mode 100644
index 00000000000..3ca252bc1e1
--- /dev/null
+++ b/ql/src/test/queries/clientpositive/union_all_empty_branch_no_tmp_dir.q
@@ -0,0 +1,87 @@
+set hive.mapred.mode=nonstrict;
+
+DROP TABLE IF EXISTS union_src_today;
+DROP TABLE IF EXISTS union_src_prev;
+DROP TABLE IF EXISTS union_output;
+
+CREATE TABLE union_src_today (id INT, val STRING) STORED AS TEXTFILE;
+CREATE TABLE union_src_prev (id INT, val STRING) STORED AS TEXTFILE;
+
+CREATE TABLE union_output (id INT, val STRING)
+PARTITIONED BY (run_dt STRING)
+STORED AS TEXTFILE;
+
+-- -----------------------------------------------------------------------
+-- Scenario A: Both branches produce rows
+-- prev has rows 1,2,3 — today has rows 3,4 — anti-join passes rows 1,2
+-- Branch 1 (anti-join): rows from prev NOT in today → ids 1,2
+-- Branch 2 (today): all rows from today → ids 3,4
+-- -----------------------------------------------------------------------
+INSERT INTO union_src_prev VALUES (1,'aaa'), (2,'bbb'), (3,'ccc');
+INSERT INTO union_src_today VALUES (3,'ccc_new'), (4,'ddd');
+
+INSERT OVERWRITE TABLE union_output PARTITION (run_dt='2024-01-02')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today;
+
+-- Partition must contain exactly the output rows (ids 1,2,3,4)
+SELECT id, val FROM union_output WHERE run_dt='2024-01-02' ORDER BY id;
+
+-- Partition directory must NOT contain any -tmp.*.moved directories
+dfs -ls
${hiveconf:hive.metastore.warehouse.dir}/union_output/run_dt=2024-01-02/;
+
+-- -----------------------------------------------------------------------
+-- Scenario B: First branch produces 0 rows (all prev rows exist in today)
+-- prev has rows 3,4 — today has rows 3,4 — anti-join returns nothing
+-- Branch 1 (anti-join): 0 rows ← triggers the bug
+-- Branch 2 (today): rows 3,4
+-- -----------------------------------------------------------------------
+TRUNCATE TABLE union_src_prev;
+TRUNCATE TABLE union_src_today;
+
+INSERT INTO union_src_prev VALUES (3,'ccc'), (4,'ddd');
+INSERT INTO union_src_today VALUES (3,'ccc_new'), (4,'ddd_new');
+
+INSERT OVERWRITE TABLE union_output PARTITION (run_dt='2024-01-03')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today;
+
+-- Partition must contain only Branch 2 rows (ids 3,4)
+SELECT id, val FROM union_output WHERE run_dt='2024-01-03' ORDER BY id;
+
+-- Partition directory must NOT contain any -tmp.*.moved directories.
+-- Before the fix, -tmp.HIVE_UNION_SUBDIR_1.moved would appear here.
+dfs -ls
${hiveconf:hive.metastore.warehouse.dir}/union_output/run_dt=2024-01-03/;
+
+-- with flatten subdirectories enabled
+set hive.tez.union.flatten.subdirectories=true;
+
+INSERT OVERWRITE TABLE union_output PARTITION (run_dt='2024-01-03')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today;
+
+-- Partition must contain only Branch 2 rows (ids 3,4)
+SELECT id, val FROM union_output WHERE run_dt='2024-01-03' ORDER BY id;
+
+-- Partition directory must NOT contain any -tmp.*.moved directories.
+-- Before the fix, -tmp.HIVE_UNION_SUBDIR_1.moved would appear here.
+dfs -ls
${hiveconf:hive.metastore.warehouse.dir}/union_output/run_dt=2024-01-03/;
+
+DROP TABLE IF EXISTS union_src_today;
+DROP TABLE IF EXISTS union_src_prev;
+DROP TABLE IF EXISTS union_output;
\ No newline at end of file
diff --git
a/ql/src/test/results/clientpositive/tez/union_all_empty_branch_no_tmp_dir.q.out
b/ql/src/test/results/clientpositive/tez/union_all_empty_branch_no_tmp_dir.q.out
new file mode 100644
index 00000000000..25cbe53d395
--- /dev/null
+++
b/ql/src/test/results/clientpositive/tez/union_all_empty_branch_no_tmp_dir.q.out
@@ -0,0 +1,251 @@
+PREHOOK: query: DROP TABLE IF EXISTS union_src_today
+PREHOOK: type: DROPTABLE
+PREHOOK: Output: database:default
+POSTHOOK: query: DROP TABLE IF EXISTS union_src_today
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Output: database:default
+PREHOOK: query: DROP TABLE IF EXISTS union_src_prev
+PREHOOK: type: DROPTABLE
+PREHOOK: Output: database:default
+POSTHOOK: query: DROP TABLE IF EXISTS union_src_prev
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Output: database:default
+PREHOOK: query: DROP TABLE IF EXISTS union_output
+PREHOOK: type: DROPTABLE
+PREHOOK: Output: database:default
+POSTHOOK: query: DROP TABLE IF EXISTS union_output
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Output: database:default
+PREHOOK: query: CREATE TABLE union_src_today (id INT, val STRING) STORED AS
TEXTFILE
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@union_src_today
+POSTHOOK: query: CREATE TABLE union_src_today (id INT, val STRING) STORED AS
TEXTFILE
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@union_src_today
+PREHOOK: query: CREATE TABLE union_src_prev (id INT, val STRING) STORED AS
TEXTFILE
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@union_src_prev
+POSTHOOK: query: CREATE TABLE union_src_prev (id INT, val STRING) STORED AS
TEXTFILE
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@union_src_prev
+PREHOOK: query: CREATE TABLE union_output (id INT, val STRING)
+PARTITIONED BY (run_dt STRING)
+STORED AS TEXTFILE
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@union_output
+POSTHOOK: query: CREATE TABLE union_output (id INT, val STRING)
+PARTITIONED BY (run_dt STRING)
+STORED AS TEXTFILE
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@union_output
+PREHOOK: query: INSERT INTO union_src_prev VALUES (1,'aaa'), (2,'bbb'),
(3,'ccc')
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@union_src_prev
+POSTHOOK: query: INSERT INTO union_src_prev VALUES (1,'aaa'), (2,'bbb'),
(3,'ccc')
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@union_src_prev
+POSTHOOK: Lineage: union_src_prev.id SCRIPT []
+POSTHOOK: Lineage: union_src_prev.val SCRIPT []
+PREHOOK: query: INSERT INTO union_src_today VALUES (3,'ccc_new'), (4,'ddd')
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@union_src_today
+POSTHOOK: query: INSERT INTO union_src_today VALUES (3,'ccc_new'), (4,'ddd')
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@union_src_today
+POSTHOOK: Lineage: union_src_today.id SCRIPT []
+POSTHOOK: Lineage: union_src_today.val SCRIPT []
+PREHOOK: query: INSERT OVERWRITE TABLE union_output PARTITION
(run_dt='2024-01-02')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today
+PREHOOK: type: QUERY
+PREHOOK: Input: default@union_src_prev
+PREHOOK: Input: default@union_src_today
+PREHOOK: Output: default@union_output@run_dt=2024-01-02
+POSTHOOK: query: INSERT OVERWRITE TABLE union_output PARTITION
(run_dt='2024-01-02')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@union_src_prev
+POSTHOOK: Input: default@union_src_today
+POSTHOOK: Output: default@union_output@run_dt=2024-01-02
+POSTHOOK: Lineage: union_output PARTITION(run_dt=2024-01-02).id EXPRESSION
[(union_src_prev)p.FieldSchema(name:id, type:int, comment:null),
(union_src_today)union_src_today.FieldSchema(name:id, type:int, comment:null), ]
+POSTHOOK: Lineage: union_output PARTITION(run_dt=2024-01-02).val EXPRESSION
[(union_src_prev)p.FieldSchema(name:val, type:string, comment:null),
(union_src_today)union_src_today.FieldSchema(name:val, type:string,
comment:null), ]
+PREHOOK: query: SELECT id, val FROM union_output WHERE run_dt='2024-01-02'
ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@union_output
+PREHOOK: Input: default@union_output@run_dt=2024-01-02
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: SELECT id, val FROM union_output WHERE run_dt='2024-01-02'
ORDER BY id
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@union_output
+POSTHOOK: Input: default@union_output@run_dt=2024-01-02
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+1 aaa
+2 bbb
+3 ccc_new
+4 ddd
+Found 2 items
+drwxr-xr-x - ### USER ### ### GROUP ### 0 ### HDFS DATE ###
hdfs://### HDFS PATH ###
+drwxr-xr-x - ### USER ### ### GROUP ### 0 ### HDFS DATE ###
hdfs://### HDFS PATH ###
+PREHOOK: query: TRUNCATE TABLE union_src_prev
+PREHOOK: type: TRUNCATETABLE
+PREHOOK: Output: default@union_src_prev
+POSTHOOK: query: TRUNCATE TABLE union_src_prev
+POSTHOOK: type: TRUNCATETABLE
+POSTHOOK: Output: default@union_src_prev
+PREHOOK: query: TRUNCATE TABLE union_src_today
+PREHOOK: type: TRUNCATETABLE
+PREHOOK: Output: default@union_src_today
+POSTHOOK: query: TRUNCATE TABLE union_src_today
+POSTHOOK: type: TRUNCATETABLE
+POSTHOOK: Output: default@union_src_today
+PREHOOK: query: INSERT INTO union_src_prev VALUES (3,'ccc'), (4,'ddd')
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@union_src_prev
+POSTHOOK: query: INSERT INTO union_src_prev VALUES (3,'ccc'), (4,'ddd')
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@union_src_prev
+POSTHOOK: Lineage: union_src_prev.id SCRIPT []
+POSTHOOK: Lineage: union_src_prev.val SCRIPT []
+PREHOOK: query: INSERT INTO union_src_today VALUES (3,'ccc_new'), (4,'ddd_new')
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@union_src_today
+POSTHOOK: query: INSERT INTO union_src_today VALUES (3,'ccc_new'),
(4,'ddd_new')
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@union_src_today
+POSTHOOK: Lineage: union_src_today.id SCRIPT []
+POSTHOOK: Lineage: union_src_today.val SCRIPT []
+PREHOOK: query: INSERT OVERWRITE TABLE union_output PARTITION
(run_dt='2024-01-03')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today
+PREHOOK: type: QUERY
+PREHOOK: Input: default@union_src_prev
+PREHOOK: Input: default@union_src_today
+PREHOOK: Output: default@union_output@run_dt=2024-01-03
+POSTHOOK: query: INSERT OVERWRITE TABLE union_output PARTITION
(run_dt='2024-01-03')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@union_src_prev
+POSTHOOK: Input: default@union_src_today
+POSTHOOK: Output: default@union_output@run_dt=2024-01-03
+POSTHOOK: Lineage: union_output PARTITION(run_dt=2024-01-03).id EXPRESSION
[(union_src_prev)p.FieldSchema(name:id, type:int, comment:null),
(union_src_today)union_src_today.FieldSchema(name:id, type:int, comment:null), ]
+POSTHOOK: Lineage: union_output PARTITION(run_dt=2024-01-03).val EXPRESSION
[(union_src_prev)p.FieldSchema(name:val, type:string, comment:null),
(union_src_today)union_src_today.FieldSchema(name:val, type:string,
comment:null), ]
+PREHOOK: query: SELECT id, val FROM union_output WHERE run_dt='2024-01-03'
ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@union_output
+PREHOOK: Input: default@union_output@run_dt=2024-01-03
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: SELECT id, val FROM union_output WHERE run_dt='2024-01-03'
ORDER BY id
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@union_output
+POSTHOOK: Input: default@union_output@run_dt=2024-01-03
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+3 ccc_new
+4 ddd_new
+Found 1 items
+drwxr-xr-x - ### USER ### ### GROUP ### 0 ### HDFS DATE ###
hdfs://### HDFS PATH ###
+PREHOOK: query: INSERT OVERWRITE TABLE union_output PARTITION
(run_dt='2024-01-03')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today
+PREHOOK: type: QUERY
+PREHOOK: Input: default@union_src_prev
+PREHOOK: Input: default@union_src_today
+PREHOOK: Output: default@union_output@run_dt=2024-01-03
+POSTHOOK: query: INSERT OVERWRITE TABLE union_output PARTITION
(run_dt='2024-01-03')
+SELECT p.id, p.val
+FROM union_src_prev p
+LEFT OUTER JOIN union_src_today t ON p.id = t.id
+WHERE t.id IS NULL
+UNION ALL
+SELECT id, val
+FROM union_src_today
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@union_src_prev
+POSTHOOK: Input: default@union_src_today
+POSTHOOK: Output: default@union_output@run_dt=2024-01-03
+POSTHOOK: Lineage: union_output PARTITION(run_dt=2024-01-03).id EXPRESSION
[(union_src_prev)p.FieldSchema(name:id, type:int, comment:null),
(union_src_today)union_src_today.FieldSchema(name:id, type:int, comment:null), ]
+POSTHOOK: Lineage: union_output PARTITION(run_dt=2024-01-03).val EXPRESSION
[(union_src_prev)p.FieldSchema(name:val, type:string, comment:null),
(union_src_today)union_src_today.FieldSchema(name:val, type:string,
comment:null), ]
+PREHOOK: query: SELECT id, val FROM union_output WHERE run_dt='2024-01-03'
ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@union_output
+PREHOOK: Input: default@union_output@run_dt=2024-01-03
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: SELECT id, val FROM union_output WHERE run_dt='2024-01-03'
ORDER BY id
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@union_output
+POSTHOOK: Input: default@union_output@run_dt=2024-01-03
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+3 ccc_new
+4 ddd_new
+Found 1 items
+-rw-rw-rw- 3 ### USER ### ### GROUP ### ### SIZE ### ### HDFS DATE
### hdfs://### HDFS PATH ###
+PREHOOK: query: DROP TABLE IF EXISTS union_src_today
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@union_src_today
+PREHOOK: Output: database:default
+PREHOOK: Output: default@union_src_today
+POSTHOOK: query: DROP TABLE IF EXISTS union_src_today
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Input: default@union_src_today
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@union_src_today
+PREHOOK: query: DROP TABLE IF EXISTS union_src_prev
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@union_src_prev
+PREHOOK: Output: database:default
+PREHOOK: Output: default@union_src_prev
+POSTHOOK: query: DROP TABLE IF EXISTS union_src_prev
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Input: default@union_src_prev
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@union_src_prev
+PREHOOK: query: DROP TABLE IF EXISTS union_output
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@union_output
+PREHOOK: Output: database:default
+PREHOOK: Output: default@union_output
+POSTHOOK: query: DROP TABLE IF EXISTS union_output
+POSTHOOK: type: DROPTABLE
+POSTHOOK: Input: default@union_output
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@union_output
