0dunay0 commented on code in PR #10673:
URL: https://github.com/apache/paimon/pull/10673#discussion_r4351892160
##########
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:
Recursive listing uses `FileIO.listFiles(root, true)`. On S3 and other
object stores, this is one flat listing of the prefix. If we prune hidden
directories during the traversal, we need one list call for each directory.
That is much slower for the deep partition layouts that this option is for.
Also, if the job can not read a directory, the clone fails with an access error
and does not copy incomplete data. So I prefer to keep the flat listing. Let me
know if you think differently.
--
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]
