content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.arrow.dataset.file;
  
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.google.common.primitives.Primitives;
  import java.io.File;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.stream.Collectors;
  import org.apache.arrow.dataset.OrcWriteSupport;
  import org.apache.arrow.dataset.ParquetWriteSupport;
  import org.apache.arrow.dataset.TextBasedWriteSupport;
  import org.apache.arrow.dataset.jni.NativeDataset;
  import org.apache.arrow.dataset.jni.NativeInstanceReleasedException;
  import org.apache.arrow.dataset.jni.NativeMemoryPool;
  import org.apache.arrow.dataset.jni.NativeScanner;
  import org.apache.arrow.dataset.jni.TestNativeDataset;
  import org.apache.arrow.dataset.scanner.ScanOptions;
  import org.apache.arrow.util.AutoCloseables;
  import org.apache.arrow.vector.FieldVector;
  import org.apache.arrow.vector.IntVector;
  import org.apache.arrow.vector.VectorLoader;
  import org.apache.arrow.vector.VectorSchemaRoot;
  import org.apache.arrow.vector.ipc.ArrowFileWriter;
  import org.apache.arrow.vector.ipc.ArrowReader;
  import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
  import org.apache.arrow.vector.types.Types;
  import org.apache.arrow.vector.types.pojo.ArrowType;
  import org.apache.arrow.vector.types.pojo.Field;
  import org.apache.arrow.vector.types.pojo.Schema;
  import org.apache.avro.generic.GenericRecord;
  import org.apache.avro.generic.GenericRecordBuilder;
  import org.apache.hadoop.fs.Path;
  import org.apache.orc.TypeDescription;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.io.TempDir;
  
  public class TestFileSystemDataset extends TestNativeDataset {
  
    @TempDir public File TMP;
  
    public static final String AVRO_SCHEMA_USER = "user.avsc";
  
    @Test
    public void testBaseParquetRead() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      ScanOptions options = new ScanOptions(100);
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(1, datum.size());
      assertEquals(2, schema.getFields().size());
      assertEquals("id", schema.getFields().get(0).getName());
      assertEquals("name", schema.getFields().get(1).getName());
      assertEquals(Types.MinorType.INT.getType(), schema.getFields().get(0).getType());
      assertEquals(Types.MinorType.VARCHAR.getType(), schema.getFields().get(1).getType());
      checkParquetReadResult(schema, writeSupport.getWrittenRecords(), datum);
  
      AutoCloseables.close(datum);
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testMultipleParquetReadFromUris() throws Exception {
      ParquetWriteSupport writeSupport1 =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
      ParquetWriteSupport writeSupport2 =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 2, "b");
      String expectedJsonUnordered = "[[1,\"a\"],[2,\"b\"]]";
  
      ScanOptions options = new ScanOptions(1);
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              new String[] {writeSupport1.getOutputURI(), writeSupport2.getOutputURI()});
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(2, datum.size());
      datum.forEach(batch -> assertEquals(1, batch.getLength()));
      checkParquetReadResult(schema, expectedJsonUnordered, datum);
  
      AutoCloseables.close(datum);
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testMultipleParquetInvalidUri() throws Exception {
      RuntimeException exc =
          assertThrows(
              RuntimeException.class,
              () ->
                  new FileSystemDatasetFactory(
                      rootAllocator(),
                      NativeMemoryPool.getDefault(),
                      FileFormat.PARQUET,
                      new String[] {"https://example.com", "file:///test/location"}));
      assertEquals("Unrecognized filesystem type in URI: https://example.com", exc.getMessage());
    }
  
    @Test
    public void testMultipleParquetMultipleFilesystemTypes() throws Exception {
      RuntimeException exc =
          assertThrows(
              RuntimeException.class,
              () ->
                  new FileSystemDatasetFactory(
                      rootAllocator(),
                      NativeMemoryPool.getDefault(),
                      FileFormat.PARQUET,
                      new String[] {"file:///test/location", "s3:///test/bucket/file"}));
      assertTrue(
          exc.getMessage()
              .startsWith(
                  "The filesystem expected a URI with one of the schemes (file) but received s3"));
    }
  
    @Test
    public void testParquetProjectSingleColumn() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      ScanOptions options = new ScanOptions(100, Optional.of(new String[] {"id"}));
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
      org.apache.avro.Schema expectedSchema = truncateAvroSchema(writeSupport.getAvroSchema(), 0, 1);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(1, schema.getFields().size());
      assertEquals("id", schema.getFields().get(0).getName());
      assertEquals(Types.MinorType.INT.getType(), schema.getFields().get(0).getType());
      assertEquals(1, datum.size());
      checkParquetReadResult(
          schema,
          Collections.singletonList(new GenericRecordBuilder(expectedSchema).set("id", 1).build()),
          datum);
  
      AutoCloseables.close(datum);
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testParquetBatchSize() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a", 2, "b", 3, "c");
  
      ScanOptions options = new ScanOptions(1);
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(3, datum.size());
      datum.forEach(batch -> assertEquals(1, batch.getLength()));
      checkParquetReadResult(schema, writeSupport.getWrittenRecords(), datum);
  
      AutoCloseables.close(datum);
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testParquetDirectoryRead() throws Exception {
      final File outputFolder = TMP;
      ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, outputFolder, 1, "a", 2, "b", 3, "c");
      ParquetWriteSupport.writeTempFile(
          AVRO_SCHEMA_USER, outputFolder, 4, "e", 5, "f", 6, "g", 7, "h");
      String expectedJsonUnordered =
          "[[1,\"a\"],[2,\"b\"],[3,\"c\"],[4,\"e\"],[5,\"f\"],[6,\"g\"],[7,\"h\"]]";
  
      ScanOptions options = new ScanOptions(new String[0], 1);
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              outputFolder.toURI().toString());
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(7, datum.size());
      datum.forEach(batch -> assertEquals(1, batch.getLength()));
      checkParquetReadResult(schema, expectedJsonUnordered, datum);
  
      AutoCloseables.close(datum);
    }
  
    @Test
    public void testEmptyProjectSelectsZeroColumns() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      ScanOptions options = new ScanOptions(100, Optional.of(new String[0]));
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
      org.apache.avro.Schema expectedSchema =
          org.apache.avro.Schema.createRecord(Collections.emptyList());
  
      assertScanBatchesProduced(factory, options);
      assertEquals(0, schema.getFields().size());
      assertEquals(1, datum.size());
      checkParquetReadResult(
          schema, Collections.singletonList(new GenericRecordBuilder(expectedSchema).build()), datum);
  
      AutoCloseables.close(datum);
    }
  
    @Test
    public void testNullProjectSelectsAllColumns() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      ScanOptions options = new ScanOptions(100, Optional.empty());
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(1, datum.size());
      assertEquals(2, schema.getFields().size());
      assertEquals("id", schema.getFields().get(0).getName());
      assertEquals("name", schema.getFields().get(1).getName());
      assertEquals(Types.MinorType.INT.getType(), schema.getFields().get(0).getType());
      assertEquals(Types.MinorType.VARCHAR.getType(), schema.getFields().get(1).getType());
      checkParquetReadResult(schema, writeSupport.getWrittenRecords(), datum);
  
      AutoCloseables.close(datum);
    }
  
    @Test
    public void testNoErrorWhenCloseAgain() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
  
      assertDoesNotThrow(
          () -> {
            NativeDataset dataset = factory.finish();
            dataset.close();
            dataset.close();
          });
  
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testErrorThrownWhenScanBatchesAgain() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      NativeDataset dataset = factory.finish();
      ScanOptions options = new ScanOptions(100);
      NativeScanner scanner = dataset.newScan(options);
      List<ArrowRecordBatch> datum = collectTaskData(scanner);
      AutoCloseables.close(datum);
      UnsupportedOperationException uoe =
          assertThrows(UnsupportedOperationException.class, scanner::scanBatches);
      assertEquals(
          "NativeScanner can only be executed once. Create a new scanner instead", uoe.getMessage());
  
      AutoCloseables.close(scanner, dataset, factory);
    }
  
    @Test
    public void testScanBatchesInOtherThread() throws Exception {
      ExecutorService executor = Executors.newSingleThreadExecutor();
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      NativeDataset dataset = factory.finish();
      ScanOptions options = new ScanOptions(100);
      NativeScanner scanner = dataset.newScan(options);
      List<ArrowRecordBatch> datum = executor.submit(() -> collectTaskData(scanner)).get();
  
      AutoCloseables.close(datum);
      AutoCloseables.close(scanner, dataset, factory);
    }
  
    @Test
    public void testErrorThrownWhenScanBatchesAfterScannerClose() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      NativeDataset dataset = factory.finish();
      ScanOptions options = new ScanOptions(100);
      NativeScanner scanner = dataset.newScan(options);
      scanner.close();
      assertThrows(NativeInstanceReleasedException.class, scanner::scanBatches);
  
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testErrorThrownWhenReadAfterNativeReaderClose() throws Exception {
      ParquetWriteSupport writeSupport =
          ParquetWriteSupport.writeTempFile(AVRO_SCHEMA_USER, TMP, 1, "a");
  
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.PARQUET,
              writeSupport.getOutputURI());
      NativeDataset dataset = factory.finish();
      ScanOptions options = new ScanOptions(100);
      NativeScanner scanner = dataset.newScan(options);
      ArrowReader reader = scanner.scanBatches();
      scanner.close();
      assertThrows(NativeInstanceReleasedException.class, reader::loadNextBatch);
  
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testBaseArrowIpcRead() throws Exception {
      File dataFile = new File(TMP, "datafile");
      Schema sourceSchema =
          new Schema(Collections.singletonList(Field.nullable("ints", new ArrowType.Int(32, true))));
      try (VectorSchemaRoot root = VectorSchemaRoot.create(sourceSchema, rootAllocator());
          FileOutputStream sink = new FileOutputStream(dataFile);
          ArrowFileWriter writer =
              new ArrowFileWriter(root, /* provider= */ null, sink.getChannel())) {
        IntVector ints = (IntVector) root.getVector(0);
        ints.setSafe(0, 0);
        ints.setSafe(1, 1024);
        ints.setSafe(2, Integer.MAX_VALUE);
        root.setRowCount(3);
        writer.start();
        writer.writeBatch();
        writer.end();
      }
  
      String arrowDataURI = dataFile.toURI().toString();
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(), NativeMemoryPool.getDefault(), FileFormat.ARROW_IPC, arrowDataURI);
      ScanOptions options = new ScanOptions(100);
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(1, datum.size());
      assertEquals(1, schema.getFields().size());
      assertEquals("ints", schema.getFields().get(0).getName());
  
      String expectedJsonUnordered = String.format("[[0],[1024],[%d]]", Integer.MAX_VALUE);
      checkParquetReadResult(schema, expectedJsonUnordered, datum);
  
      AutoCloseables.close(datum);
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testBaseOrcRead() throws Exception {
      String dataName = "test-orc";
      String basePath = TMP.getAbsolutePath();
  
      TypeDescription orcSchema = TypeDescription.fromString("struct<ints:int>");
      Path path = new Path(basePath, dataName);
      OrcWriteSupport.writeTempFile(
          orcSchema, path, new Integer[] {Integer.MIN_VALUE, Integer.MAX_VALUE});
  
      String orcDatasetUri = new File(basePath, dataName).toURI().toString();
      FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(), NativeMemoryPool.getDefault(), FileFormat.ORC, orcDatasetUri);
      ScanOptions options = new ScanOptions(100);
      Schema schema = inferResultSchemaFromFactory(factory, options);
      List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
  
      assertScanBatchesProduced(factory, options);
      assertEquals(1, datum.size());
      assertEquals(1, schema.getFields().size());
      assertEquals("ints", schema.getFields().get(0).getName());
  
      String expectedJsonUnordered = "[[2147483647], [-2147483648]]";
      checkParquetReadResult(schema, expectedJsonUnordered, datum);
  
      AutoCloseables.close(datum);
      AutoCloseables.close(factory);
    }
  
    @Test
    public void testBaseCsvRead() throws Exception {
      TextBasedWriteSupport writeSupport =
          TextBasedWriteSupport.writeTempFile(
              TMP, ".csv", "Name,Language", "Juno,Java", "Peter,Python", "Celin,C++");
      String expectedJsonUnordered =
          "[[\"Juno\", \"Java\"], [\"Peter\", \"Python\"], [\"Celin\", \"C++\"]]";
      ScanOptions options = new ScanOptions(100);
      try (FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.CSV,
              writeSupport.getOutputURI())) {
        List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
        Schema schema = inferResultSchemaFromFactory(factory, options);
  
        assertScanBatchesProduced(factory, options);
        assertEquals(1, datum.size());
        assertEquals(2, schema.getFields().size());
        assertEquals("Name", schema.getFields().get(0).getName());
  
        checkParquetReadResult(schema, expectedJsonUnordered, datum);
  
        AutoCloseables.close(datum);
      }
    }
  
    @Test
    public void testBaseJsonRead() throws Exception {
      TextBasedWriteSupport writeSupport =
          TextBasedWriteSupport.writeTempFile(
              TMP,
              ".json",
              "{\"Type\": \"Compiled\", \"Language\": \"Java\"}",
              "{\"Type\": \"Interpreted\", \"Language\": \"Python\"}");
      String expectedJsonUnordered = "[[\"Compiled\", \"Java\"], " + "[\"Interpreted\", \"Python\"]]";
      ScanOptions options = new ScanOptions(100);
      try (FileSystemDatasetFactory factory =
          new FileSystemDatasetFactory(
              rootAllocator(),
              NativeMemoryPool.getDefault(),
              FileFormat.JSON,
              writeSupport.getOutputURI())) {
        List<ArrowRecordBatch> datum = collectResultFromFactory(factory, options);
        Schema schema = inferResultSchemaFromFactory(factory, options);
  
        assertScanBatchesProduced(factory, options);
        assertEquals(1, datum.size());
        assertEquals(2, schema.getFields().size());
        assertEquals("Type", schema.getFields().get(0).getName());
        assertEquals("Language", schema.getFields().get(1).getName());
  
        checkParquetReadResult(schema, expectedJsonUnordered, datum);
  
        AutoCloseables.close(datum);
      }
    }
  
    private void checkParquetReadResult(
        Schema schema, String expectedJson, List<ArrowRecordBatch> actual) throws IOException {
      final ObjectMapper json = new ObjectMapper();
      final Set<?> expectedSet = json.readValue(expectedJson, Set.class);
      final Set<List<Object>> actualSet = new HashSet<>();
      final int fieldCount = schema.getFields().size();
      try (VectorSchemaRoot vsr = VectorSchemaRoot.create(schema, rootAllocator())) {
        VectorLoader loader = new VectorLoader(vsr);
        for (ArrowRecordBatch batch : actual) {
          loader.load(batch);
          int batchRowCount = vsr.getRowCount();
          for (int i = 0; i < batchRowCount; i++) {
            List<Object> row = new ArrayList<>();
            for (int j = 0; j < fieldCount; j++) {
              Object object = vsr.getVector(j).getObject(i);
              if (Primitives.isWrapperType(object.getClass())) {
                row.add(object);
              } else {
                row.add(object.toString());
              }
            }
            actualSet.add(row);
          }
        }
      }
      assertEquals(
          expectedSet,
          actualSet,
          "Mismatched data read from Parquet, actual: " + json.writeValueAsString(actualSet) + ";");
    }
  
    private void checkParquetReadResult(
        Schema schema, List<GenericRecord> expected, List<ArrowRecordBatch> actual) {
      assertEquals(expected.size(), actual.stream().mapToInt(ArrowRecordBatch::getLength).sum());
      final int fieldCount = schema.getFields().size();
      ArrayList<GenericRecord> expectedRemovable = new ArrayList<>(expected);
      try (VectorSchemaRoot vsr = VectorSchemaRoot.create(schema, rootAllocator())) {
        VectorLoader loader = new VectorLoader(vsr);
        for (ArrowRecordBatch batch : actual) {
          assertEquals(fieldCount, batch.getNodes().size());
          loader.load(batch);
          int batchRowCount = vsr.getRowCount();
          for (int i = 0; i < fieldCount; i++) {
            FieldVector vector = vsr.getVector(i);
            for (int j = 0; j < batchRowCount; j++) {
              Object object = vector.getObject(j);
              Object expectedObject = expectedRemovable.get(j).get(i);
              assertEquals(Objects.toString(expectedObject), Objects.toString(object));
            }
          }
          for (int i = 0; i < batchRowCount; i++) {
            expectedRemovable.remove(0);
          }
        }
        assertTrue(expectedRemovable.isEmpty());
      }
    }
  
    private org.apache.avro.Schema truncateAvroSchema(
        org.apache.avro.Schema schema, int from, int to) {
      List<org.apache.avro.Schema.Field> fields = schema.getFields().subList(from, to);
      return org.apache.avro.Schema.createRecord(
          fields.stream()
              .map(
                  f ->
                      new org.apache.avro.Schema.Field(
                          f.name(), f.schema(), f.doc(), f.defaultVal(), f.order()))
              .collect(Collectors.toList()));
    }
  }
  