[destination-postgres] Internal table indexes are duplicated leading to a misuse of diskspace

[walker-philips]

Connector Name

destination-postgres

Connector Version

2.4.4

What step the error happened?

Other

Relevant information

All postgres airbyte internal tables are being overindexed. This overindexing is A) unnecessary B) wastes disk space.

I am seeing four indexes being created on these tables:
<stream_name>_airbyte_tmp_extracted_at USING btree (_airbyte_extracted_at)
<stream_name>_airbyte_tmp_loaded_at USING btree (_airbyte_loaded_at, _airbyte_extracted_at)
<stream_name>_airbyte_tmp_pkey USING btree (_airbyte_raw_id)
<stream_name>_airbyte_tmp_raw_id USING btree (_airbyte_raw_id)

The _pkey and _raw_id are indexing on the same column.

SELECT
  c.relname,
  (pg_stat_file(pg_relation_filepath(c.oid))).modification AS file_mtime
FROM pg_class c
WHERE c.relkind = 'i'
  AND (c.relname iLIKE '%_raw_id'
	or c.relname iLIKE '%_pkey')
	order by 2 desc

For our instance shows that both indexes are being actively generated so neither appears to be remnant indexes from an earlier postgres version.
Our database monitoring tool is indicating over 700 duplicate indexes for our moderate deployment of Airbyte.

Recommended solution:
I think either index can be dropped.
If pkey was intended for use with a Dedupe sync, then the index creation is not correctly doing that.

Relevant log output

Contribute

• Yes, I want to contribute

[walker-philips]

I am specifically looking at these two blocks of code as the potential culprits. I am not super familiar with java or dslcontext. The "pkey" index creation is still a mystery to me but I am assuming it is the dslcontext.

From what I can tell, pkey indexes are ironically not created for incremental + deduped syncs which ask for a primary key.
Full refresh overwrite gets both "pkey" and "raw_id" indexes.

class PostgresSqlOperations(useDropCascade: Boolean) : JdbcSqlOperations() {
    private val dropTableQualifier: String = if (useDropCascade) "CASCADE" else ""
    override fun postCreateTableQueries(schemaName: String?, tableName: String?): List<String> {
        return if (isDestinationV2) {
            java.util.List.of( // the raw_id index _could_ be unique (since raw_id is a UUID)
                // but there's no reason to do that (because it's a UUID :P )
                // and it would just slow down inserts.
                // also, intentionally don't specify the type of index (btree, hash, etc). Just use
                // the default.
                "CREATE INDEX IF NOT EXISTS " +
                    tableName +
                    "_raw_id" +
                    " ON " +
                    schemaName +
                    "." +
                    tableName +
                    "(_airbyte_raw_id)",
                "CREATE INDEX IF NOT EXISTS " +
                    tableName +
                    "_extracted_at" +
                    " ON " +
                    schemaName +
                    "." +
                    tableName +
                    "(_airbyte_extracted_at)",
                "CREATE INDEX IF NOT EXISTS " +
                    tableName +
                    "_loaded_at" +
                    " ON " +
                    schemaName +
                    "." +
                    tableName +
                    "(_airbyte_loaded_at, _airbyte_extracted_at)"
            )
        } else {
            emptyList()
        }
    }

and

override fun createTable(stream: StreamConfig, suffix: String, force: Boolean): Sql {
        val statements: MutableList<Sql> = ArrayList()
        val finalTableName = DSL.name(stream.id.finalNamespace, stream.id.finalName + suffix)

        statements.add(super.createTable(stream, suffix, force))

        if (stream.postImportAction == ImportType.DEDUPE) {
            // An index for our ROW_NUMBER() PARTITION BY pk ORDER BY cursor, extracted_at function
            val pkNames =
                stream.primaryKey.stream().map { pk: ColumnId -> DSL.quotedName(pk.name) }.toList()
            statements.add(
                of(
                    dslContext
                        .createIndex()
                        .on(
                            finalTableName,
                            Stream.of<Stream<Name>>(
                                    pkNames
                                        .stream(), // if cursor is present, then a stream containing
                                    // its name
                                    // but if no cursor, then empty stream
                                    stream.cursor.stream().map<Name> { cursor: ColumnId ->
                                        DSL.quotedName(cursor.name)
                                    },
                                    Stream.of<Name>(
                                        DSL.name(JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT)
                                    )
                                )
                                .flatMap<Name>(Function.identity<Stream<Name>>())
                                .toList()
                        )
                        .sql
                )
            )
        }
        statements.add(
            of(
                dslContext
                    .createIndex()
                    .on(finalTableName, DSL.name(JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT))
                    .getSQL()
            )
        )

        statements.add(
            of(
                dslContext
                    .createIndex()
                    .on(finalTableName, DSL.name(JavaBaseConstants.COLUMN_NAME_AB_RAW_ID))
                    .getSQL()
            )
        )

        return concat(statements)
    }