danny0405 commented on code in PR #18776:
URL: https://github.com/apache/hudi/pull/18776#discussion_r4059111321
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -137,7 +138,13 @@ protected void initFileWriter(Path outPutFile,
CompressionCodecName newCodecName
ParquetFileWriter.Mode writerMode = ParquetFileWriter.Mode.CREATE;
writer = new ParquetFileWriter(HadoopOutputFile.fromPath(outPutFile,
conf), schema, writerMode, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT,
DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
DEFAULT_STATISTICS_TRUNCATE_LENGTH,
ParquetProperties.DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED);
- writer.start();
+ try {
+ writer.start();
+ } catch (Exception e) {
+ closeParquetFileWriterQuietly(writer);
Review Comment:
Done: `closeParquetFileWriterQuietly` captures the current writer and
immediately clears the field before attempting cleanup. The close tests assert
that the field is cleared even when finalization or cleanup fails.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -505,32 +541,66 @@ private void addNullColumn(ColumnDescriptor descriptor,
long totalChunkValues, E
new ColumnChunkPageWriteStore(compressor, newSchema,
props.getAllocator(), props.getColumnIndexTruncateLength(),
props.getPageWriteChecksumEnabled(), null, numBlocksRewritten);
ColumnWriteStore cStore = props.newColumnWriteStore(newSchema, cPageStore);
ColumnWriter cWriter = cStore.getColumnWriter(descriptor);
- int dMax = descriptor.getMaxDefinitionLevel();
-
- for (int i = 0; i < totalChunkValues; i++) {
- int rlvl = 0;
- int dlvl = 0;
- if (dlvl == dMax) {
- // since we checked ether optional or repeated, dlvl should be > 0
- if (dlvl == 0) {
- throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
- }
- // we just write one null for the whole list at the top level,
- // instead of nullify the elements in the list one by one
- if (rlvl == 0) {
- cWriter.writeNull(rlvl, dlvl - 1);
+ Throwable failure = null;
+ try {
+ int dMax = descriptor.getMaxDefinitionLevel();
+
+ for (int i = 0; i < totalChunkValues; i++) {
+ int rlvl = 0;
+ int dlvl = 0;
+ if (dlvl == dMax) {
+ // since we checked ether optional or repeated, dlvl should be > 0
+ if (dlvl == 0) {
+ throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
+ }
+ // we just write one null for the whole list at the top level,
+ // instead of nullify the elements in the list one by one
+ if (rlvl == 0) {
+ cWriter.writeNull(rlvl, dlvl - 1);
+ }
+ } else {
+ cWriter.writeNull(rlvl, dlvl); // 因为repeatition
level没有重复所以后面都是以0在第一层，definition level是字段path的第0层
Review Comment:
Replaced it with an English comment explaining the zero repetition and
definition levels.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -505,32 +541,66 @@ private void addNullColumn(ColumnDescriptor descriptor,
long totalChunkValues, E
new ColumnChunkPageWriteStore(compressor, newSchema,
props.getAllocator(), props.getColumnIndexTruncateLength(),
props.getPageWriteChecksumEnabled(), null, numBlocksRewritten);
ColumnWriteStore cStore = props.newColumnWriteStore(newSchema, cPageStore);
ColumnWriter cWriter = cStore.getColumnWriter(descriptor);
- int dMax = descriptor.getMaxDefinitionLevel();
-
- for (int i = 0; i < totalChunkValues; i++) {
- int rlvl = 0;
- int dlvl = 0;
- if (dlvl == dMax) {
- // since we checked ether optional or repeated, dlvl should be > 0
- if (dlvl == 0) {
- throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
- }
- // we just write one null for the whole list at the top level,
- // instead of nullify the elements in the list one by one
- if (rlvl == 0) {
- cWriter.writeNull(rlvl, dlvl - 1);
+ Throwable failure = null;
+ try {
+ int dMax = descriptor.getMaxDefinitionLevel();
+
+ for (int i = 0; i < totalChunkValues; i++) {
+ int rlvl = 0;
+ int dlvl = 0;
+ if (dlvl == dMax) {
+ // since we checked ether optional or repeated, dlvl should be > 0
+ if (dlvl == 0) {
+ throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
+ }
+ // we just write one null for the whole list at the top level,
+ // instead of nullify the elements in the list one by one
+ if (rlvl == 0) {
+ cWriter.writeNull(rlvl, dlvl - 1);
+ }
+ } else {
+ cWriter.writeNull(rlvl, dlvl); // 因为repeatition
level没有重复所以后面都是以0在第一层，definition level是字段path的第0层
}
- } else {
- cWriter.writeNull(rlvl, dlvl); // 因为repeatition
level没有重复所以后面都是以0在第一层，definition level是字段path的第0层
+ cStore.endRecord();
}
- cStore.endRecord();
- }
- cStore.flush();
- cPageStore.flushToFileWriter(writer);
+ cStore.flush();
+ cPageStore.flushToFileWriter(writer);
+ } catch (IOException | RuntimeException e) {
+ failure = e;
+ throw e;
+ } finally {
+ closeColumnWriters(cStore, cWriter, failure);
+ }
+ }
- cStore.close();
- cWriter.close();
+ private void closeColumnWriters(ColumnWriteStore cStore, ColumnWriter
cWriter, Throwable failure) {
+ RuntimeException closeException = null;
+ try {
+ if (cStore != null) {
+ cStore.close();
+ }
+ } catch (RuntimeException re) {
+ closeException = re;
+ }
+ try {
+ if (cWriter != null) {
+ cWriter.close();
+ }
+ } catch (RuntimeException re) {
+ if (closeException == null) {
+ closeException = re;
+ } else {
+ closeException.addSuppressed(re);
+ }
+ }
+ if (closeException != null) {
+ if (failure != null) {
+ failure.addSuppressed(closeException);
+ } else {
+ throw closeException;
+ }
+ }
Review Comment:
`cWriter` is a method-local variable, so clearing it would not change the
caller's state. I simplified this path to try-with-resources using close-method
adapters for the column store and writer. Java now handles cleanup and
suppressed exceptions, and the custom `closeColumnWriters` helper has been
removed.
--
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]
