JingsongLi commented on code in PR #10673:
URL: https://github.com/apache/paimon/pull/10673#discussion_r4345687791
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/CloneHiveSchemaFunction.java:
##########
@@ -147,6 +162,12 @@ public void processElement(
"existed paimon table '%s' is not a
FileStoreTable, but a %s",
tuple.f1, existedTable.getClass().getName()));
checkCompatible(schema, (FileStoreTable) existedTable);
+ if (copyFiles) {
Review Comment:
[P1] Preserve ordered partition identity in the file-copy path
This branch accepts an existing target whose partition-key set matches Hive
but whose order differs. `ListCloneFilesFunction` supplies the target partition
type, while Hive partition values are encoded positionally in their source
order. I reproduced Hive `(p1,p2)=(A,B)` into a target partitioned by
`(p2,p1)`: cloning with `copy_files=true` returned `(p1,p2)=(B,A)`. With an
existing `keep` row in target `(B,A)`, the clone also overwrote that unrelated
partition; the only result was the mispartitioned source row. ORC/Parquet newly
reach this path through the added flag. Require identical partition-key order
for copying, or remap values by key name before filtering, path creation and
overwrite.
##########
paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/clone/HiveTableCloneExtractor.java:
##########
@@ -177,20 +225,85 @@ private static List<HivePartitionFiles>
listFromPureHiveTable(
valueSetters,
defaultPartitionName);
if (predicate == null || predicate.test(partitionRow)) {
- results.add(listFiles(fileIO, location, partitionRow,
format));
+ checkPartitionColumns(identifier, sourceTable, partition);
+ results.add(listFiles(fileIO, location, partitionRow,
format, recursive));
}
}
+ if (recursive) {
+ checkNoSharedFiles(identifier, results);
+ }
return results;
}
}
- private static HivePartitionFiles listFiles(
- FileIO fileIO, String location, BinaryRow partition, String
format) throws IOException {
- List<FileStatus> fileStatuses =
- Arrays.stream(fileIO.listStatus(new Path(location)))
- .filter(s -> !s.isDir())
- .filter(HIDDEN_PATH_FILTER)
- .collect(Collectors.toList());
+ /**
+ * Copied files are read with the table columns, so a partition must not
declare a column of the
+ * table with another type. Columns missing from a partition are read as
null.
+ */
+ private static void checkPartitionColumns(
+ Identifier identifier, Table table, Partition partition) {
+ if (!partition.getSd().isSetCols()) {
+ return;
+ }
+ Map<String, String> tableTypes = new HashMap<>();
+ for (FieldSchema column : table.getSd().getCols()) {
+ tableTypes.put(column.getName(), column.getType());
+ }
+ for (FieldSchema column : partition.getSd().getCols()) {
+ String tableType = tableTypes.get(column.getName());
+ if (tableType != null &&
!tableType.equalsIgnoreCase(column.getType())) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Can not copy the files of partition %s of
table %s: column '%s' has type '%s' in the partition but '%s' in the table.",
+ partition.getValues(),
+ identifier,
+ column.getName(),
+ column.getType(),
+ tableType));
+ }
+ }
+ }
+
+ /**
+ * Listing recursively, a partition located inside another partition would
have its files cloned
+ * twice.
+ */
+ @VisibleForTesting
+ static void checkNoSharedFiles(Identifier identifier,
List<HivePartitionFiles> partitions) {
+ Map<Path, BinaryRow> owners = new HashMap<>();
+ for (HivePartitionFiles partitionFiles : partitions) {
+ for (Path path : partitionFiles.paths()) {
+ BinaryRow owner = owners.putIfAbsent(path,
partitionFiles.partition());
+ if (owner != null) {
+ throw new IllegalStateException(
+ String.format(
+ "File %s of table %s is listed by more
than one partition, "
+ + "check the partition locations
or disable recursive listing.",
+ path, identifier));
+ }
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static HivePartitionFiles listFiles(
+ FileIO fileIO, String location, BinaryRow partition, String
format, boolean recursive)
+ throws IOException {
+ Path root = new Path(location);
+ List<FileStatus> fileStatuses;
+ if (recursive) {
+ fileStatuses =
+ Arrays.stream(fileIO.listFiles(root, true))
Review Comment:
[P2] Prune hidden directories before descending into them
`fileIO.listFiles(root, true)` traverses the entire tree before
`isInHiddenDirectory` can exclude staging paths. A readable partition
containing a private `_temporary` directory therefore fails the clone on an
access error, even though that directory is advertised as skipped. A
deterministic FileIO ACL fixture confirmed ordinary recursive listing and
denied shallow listing succeed, while denied recursive listing enters
`_temporary` and throws before this filter. Exclude hidden directories during
traversal rather than after materializing all files. No HDFS cluster was used
for this fixture.
##########
paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/clone/HiveTableCloneExtractor.java:
##########
@@ -113,7 +147,13 @@ public Map<String, String> extractOptions(Table table) {
}
String format = parseFormat(table);
- if (supportCloneSplits(format)) {
+ if (copyFiles && !COPY_FILES_FORMATS.contains(format)) {
Review Comment:
[P1] Reject transactional Hive ORC tables before raw copying
The format-only check accepts full ACID tables (`transactional=true`,
`transactional_properties=default`). I verified that the real preflight, schema
conversion and recursive listing admit their `base_*`, `delta_*` and
`delete_delta_*` files. These files contain transaction-event envelopes and
require valid-write-ID selection and delete merging; copying all of them as
ordinary Paimon ORC files cannot preserve the Hive table view. A real ORC
fixture with the standard six-field event schema also makes the current Paimon
logical-row reader fail with `ArrayIndexOutOfBoundsException`. Reject
transactional sources before target creation unless an ACID-aware conversion
path is implemented. The metadata transport in the admission fixture was a
fixed local metastore proxy; selector/preflight/schema/listing and ORC reads
used the branch implementations.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
