[core] Support ignore corrupt or lost files during read

Author: zhangyongxiang.alpha

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -129,8 +129,7 @@ public RawFileSplitRead newBatchRawFileRead() {
                 valueType,
                 FileFormatDiscover.of(options),
                 pathFactory(),
-                options.fileIndexReadEnabled(),
-                false);
+                options);
     }
 
     public KeyValueFileReaderFactory.Builder newReaderFactoryBuilder() {

paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java

@@ -52,10 +52,10 @@ public ChainKeyValueFileReaderFactory(
             RowType valueType,
             FormatReaderMapping.Builder formatReaderMappingBuilder,
             DataFilePathFactory pathFactory,
-            long asyncThreshold,
             BinaryRow partition,
             DeletionVector.Factory dvFactory,
-            ChainReadContext chainReadContext) {
+            ChainReadContext chainReadContext,
+            CoreOptions coreOptions) {
         super(
                 fileIO,
                 schemaManager,
@@ -64,9 +64,9 @@ public ChainKeyValueFileReaderFactory(
                 valueType,
                 formatReaderMappingBuilder,
                 pathFactory,
-                asyncThreshold,
                 partition,
-                dvFactory);
+                dvFactory,
+                coreOptions);
         this.chainReadContext = chainReadContext;
         CoreOptions options = new CoreOptions(schema.options());
         this.currentBranch = options.branch();
@@ -132,10 +132,10 @@ public ChainKeyValueFileReaderFactory build(
                     wrapped.readValueType,
                     builder,
                     wrapped.pathFactory.createChainReadDataFilePathFactory(chainReadContext),
-                    wrapped.options.fileReaderAsyncThreshold().getBytes(),
                     partition,
                     dvFactory,
-                    chainReadContext);
+                    chainReadContext,
+                    wrapped.options);
         }
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java

@@ -319,5 +319,9 @@ protected FormatReaderMapping.Builder formatReaderMappingBuilder(
         public FileIO fileIO() {
             return fileIO;
         }
+
+        public CoreOptions options() {
+            return options;
+        }
     }
 }

paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java

@@ -195,14 +195,17 @@ protected RecordReader<InternalRow> createReader(FormatDataSplit dataSplit) thro
             return new DataFileRecordReader(
                     readType(),
                     reader,
+                    options.scanIgnoreCorruptFile(),
+                    options.scanIgnoreLostFile(),
                     null,
                     null,
                     PartitionUtils.create(partitionMapping, dataSplit.partition()),
                     false,
                     null,
                     0,
                     Collections.emptyMap(),
-                    null);
+                    null,
+                    formatReaderContext.filePath());
         } catch (Exception e) {
             FileUtils.checkExists(formatReaderContext.fileIO(), formatReaderContext.filePath());
             throw e;