How to modify the sink config to use Position Delete

[ajosegun]

When using the current configuration to create Iceberg tables on GCP, the table is created with equality delete files by default. This prevents Pyiceberg from reading the tables due to a lack of support for equality delete files. (I get Pyiceberg does not support reading equality delete files when I try to read an Iceberg table after a change has been made in the source table)

Environment

• Library: Pyiceberg
• Platform: GCP
• Configuration: provided below

Current Behavior

• Table creation results in equality delete files
• Attempting to read the table throws the error: "Pyiceberg does not support reading equality delete files"

Reproducibility

1. Create Iceberg table using current configuration
2. Make changes (update/delete) in the source database/table
3. Attempt to read the table with Pyiceberg

Question

1. How can I modify the sink configuration to use position delete?
2. Is there a way to convert equality delete files to position delete files?

        "connector.class": "io.tabular.iceberg.connect.IcebergSinkConnector",
        "tasks.max": "2",
        "iceberg.tables.dynamic-enabled": "true",
        "iceberg.upsert": "true",
        "iceberg.upsert-keep-deletes": "true",
        "iceberg.tables.auto-create-enabled": "true",
        "iceberg.file-format": "PARQUET",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "iceberg.catalog.jdbc.password": "XXXX",
        "iceberg.catalog.jdbc.driver": "org.postgresql.Driver",
        "iceberg.tables.route-field": "__source_table",
        "iceberg.delete-files-on-table-drop": "true",
        "iceberg.catalog.io-impl": "org.apache.iceberg.gcp.gcs.GCSFileIO",
        "iceberg.control.commit.interval-ms": "60000",
        "iceberg.catalog.uri": "jdbc:postgresql://XXX.XXX.XXX.XXX/XXXXX",
        "topics.regex": "cdc\\.XXXXX\\.XXXXXX.*",
        "iceberg.catalog": "XXXXXX",
        "key.converter.schemas.enable": "false",
        "iceberg.catalog.gcs.credentials.config": "{****}",
        "value.converter.schemas.enable": "false",
        "iceberg.tables.cdc-field": "__op",
        "iceberg.catalog.jdbc.user": "******",
        "iceberg.catalog.jdbc.table-namespace": "public",
        "iceberg.catalog.warehouse": "gs://k*****3/warehouse",
        "iceberg.catalog.type": "jdbc",
        "iceberg.write.format.default": "parquet",
        "iceberg.tables.default-id-columns": "id",
        "iceberg.table.tpch_10g_tst7.customer.id-columns": "c_custkey",
        "iceberg.table.tpch_10g_tst7.nation.id-columns": "n_nationkey",
        "iceberg.table.tpch_10g_tst7.orders.id-columns": "o_orderkey",
        "iceberg.table.tpch_10g_tst7.part.id-columns": "p_partkey",
        "iceberg.table.tpch_10g_tst7.partsupp.id-columns": "ps_partkey,ps_suppkey",
        "iceberg.table.tpch_10g_tst7.region.id-columns": "r_regionkey",
        "iceberg.table.tpch_10g_tst7.supplier.id-columns": "s_suppkey",
        "iceberg.table.tpch_10g_tst7.lineitem.id-columns": "l_partkey"
    }```