public GenericBaseTaskWriter buildBaseWriter() {
      writeBasePreconditions();
      FileFormat fileFormat =
          FileFormat.valueOf(
              (table
                  .properties()
                  .getOrDefault(
                      TableProperties.BASE_FILE_FORMAT, TableProperties.BASE_FILE_FORMAT_DEFAULT)
                  .toUpperCase(Locale.ENGLISH)));
      long fileSizeBytes;
      if (this.targetFileSize > 0) {
        fileSizeBytes = this.targetFileSize;
      } else {
        fileSizeBytes =
            PropertyUtil.propertyAsLong(
                table.properties(),
                TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
                TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
      }

      long mask =
          PropertyUtil.propertyAsLong(
                  table.properties(),
                  TableProperties.BASE_FILE_INDEX_HASH_BUCKET,
                  TableProperties.BASE_FILE_INDEX_HASH_BUCKET_DEFAULT)
              - 1;
      return new GenericBaseTaskWriter(
          fileFormat,
          new GenericAppenderFactory(base.schema(), table.spec()).setAll(table.properties()),
          new CommonOutputFileFactory(
              base.location(),
              table.spec(),
              fileFormat,
              table.io(),
              base.encryption(),
              partitionId,
              taskId,
              transactionId),
          table.io(),
          fileSizeBytes,
          mask,
          base.schema(),
          table.spec(),
          primaryKeySpec,
          orderedWriter);
    }
Builder to create writers for {@link KeyedTable} writting {@link Record}.
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/GenericTaskWriters.java