DuckDB adds Ducklake and Postgres. Cannot generate parquet after a while

[DSir-ai]

Using Java language

DuckDB adds Ducklake and Postgres.
I'm using Druid to manage connections in Java. Initially, I was able to write metadata to Postgres and generate Parquet files on disk, but after a while, the generation stopped. What's the cause?

·····

The technology stack is: Spring Boot + MyBatis + DruidSource + DuckDB + DuckLake + Postgres.

DruidSource is used to manage DuckDB. DuckDB connections are managed with a data connection pool.

At the beginning of the project, I could generate Parquet files using INSERT INTO statements and view the corresponding metadata in Postgres.

However, after a few hours, the corresponding metadata in Postgres could no longer be viewed, but there were no errors.

I simulated queries using two machines.

Initially, the two machines could share data for queries because the queries were performed through Postgres, and the files were written to disk, generating Parquet files.

However, after a few hours, when one machine wrote data, the Parquet files could not be generated, and the other machine could not read it. I suspect that the data was being written directly to the memory of the current machine.

What is causing this? Is the Postgres connection to DuckDB disconnected?

How can this be avoided?

····

Sample Code :
DuckDbContextHolder.java

[staticlibs]

Hi, thanks for the report!

I suspect that the data was being written directly to the memory of the current machine

One reason for that may be the disconnection of the attached DuckLake instance. This can happen when the connection pooling is involved, when the last connection to :memory: DB is closed - the in-memory instance is destroyed and ATTACH needs to be run again on reconnect. It is possible to prevent this by using jdbc_pin_db=true option (ref) in URL or in connection properties:

Properties config = new Properties();
config.put(DuckDBDriver.JDBC_PIN_DB, true);
DriverManager.getConnection(url, config);

When DuckLake "direct attach" is used (url with jdbc:duckdb:ducklake: prefix), then jdbc_pin_db is enabled automatically (and additionally the result set streaming is enabled).

[DSir-ai]

When using ducklake,

I see that the JDBC_PIN_DB property is enabled in the source code,

but in the actual production environment, after a period of time,

the data cannot be written into postgres and only exists in the local memory.

What is the reason?

Or is it that every time a request is made, it needs to be attached, but every time it is attached, except for the first initialization, it will report an error afterwards.

[staticlibs]

@DSir-ai

ATTACH IF NOT EXISTS or direct attach using jdbc:duckdb:ducklake:... url may help here.

[DSir-ai]

@staticlibs

I am currently using the MyBatis interceptor to add a layer of attach to connect to Postgres each time I execute it.

Currently, I see no disconnection or unresponsiveness, and the data is only inserted into the memory. Parquet files can be generated on disk.

However, this approach is not very elegant. Is there any more elegant approach? Please advise.

[staticlibs]

@DSir-ai

I am currently using the MyBatis interceptor to add a layer of attach to connect to Postgres each time I execute it.

With jdbc_pin_db, if you know in front all the DBs to attach to, then you can call ATTACH once in the initial connection - DBs will stay attached.

the data is only inserted into the memory. Parquet files can be generated on disk.
However, this approach is not very elegant. Is there any more elegant approach? Please advise.

This part is not clear, can you provide a runnable example?

[DSir-ai]

@staticlibs
When I initialized the database connection, I loaded ducklake, loadpostgres and attach postgres.

But after a while, maybe an hour or so, it will be interrupted.

And there is no reminder. The program is still available, and all the data is stored in the memory.

My current solution is to attach once before each sql operation.

But if the connection is not interrupted, it will directly report an error.

If the connection is interrupted, it will reconnect.

I have tested it for a few days and have not found the problem of data being stored in memory without generating parquet files.

for example:
The current instance is OK. If you remove the interception of mybatis, it is not OK to add it only once when connecting.

1.mybatis interceptor:

@Slf4j
@Component
@Intercepts({
        @Signature(type = StatementHandler.class,
                method = "prepare",
                args = {Connection.class, Integer.class})
})
@RequiredArgsConstructor
public class SessionPreOperationInterceptor implements Interceptor {
    private final FsoConfig fsoConfig;

    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        // 1. Get the target object and method parameters
        Object[] args = invocation.getArgs();

        // 2. Get connection parameters
        Connection connection = (Connection) args[0];

        log.info("[duckdb-interceptor] Database connection obtained, performing pre-operations...");
        if (connection.getMetaData().getURL().contains("jdbc:duckdb:")) {
            try (Statement stmt = connection.createStatement()) {
                String tenantId = TenantContext.get().getTenantId();
                String attachSQL = String.format("ATTACH '%s' as tenant_lake (DATA_PATH '%s')", fsoConfig.getDuckdb().getV2().getDuckLakeAttach(), fsoConfig.completeDuckdbV2DataPath(tenantId));
                stmt.addBatch(attachSQL);
                stmt.addBatch("USE tenant_lake");

                try {
                    stmt.executeBatch();
                } catch (Exception e) {
                    log.info("[duckdb-interceptor] Connected");
                }
            }
        }

        // 3. Continue to execute the original process
        return invocation.proceed();
    }
}

2.Datasource create

@Slf4j
@Service
@RequiredArgsConstructor
public class DuckDbContextHolder {
    private final FsoConfig fsoConfig;
    private final DataSource dataSource;
    private static final String DUCKDB = "duckdb_";
    private static final String DUCKDB_DRIVER_CLASS_NAME = "org.duckdb.DuckDBDriver";
    private static final String DUCKDB_URL = "jdbc:duckdb:";
    private static final String BATCH_SQL_1 = "INSTALL ducklake;";
    private static final String BATCH_SQL_2 = "LOAD ducklake;";
    private static final String BATCH_SQL_3 = "INSTALL postgres;";
    private static final String BATCH_SQL_4 = "LOAD postgres;";
    private static final String TEST_SQL = "SELECT 1";

    /**
     * Create a Druid data source
     */
    public DataSource create() {
        DruidDataSource dds = new DruidDataSource();
        dds.setDriverClassName(DUCKDB_DRIVER_CLASS_NAME);
        dds.setUrl(DUCKDB_URL);
        dds.setMinIdle(fsoConfig.getDuckdb().getV2().getMinIdle());
        dds.setMaxActive(fsoConfig.getDuckdb().getV2().getMaxActive());
        dds.setMaxWait(fsoConfig.getDuckdb().getV2().getMaxWait());
        dds.setMaxEvictableIdleTimeMillis(fsoConfig.getDuckdb().getV2().getMaxEvictableIdleTimeMillis());
        dds.setLogAbandoned(true);
        dds.setBreakAfterAcquireFailure(true);
        dds.setConnectionErrorRetryAttempts(fsoConfig.getDuckdb().getV2().getConnectionErrorRetryAttempts());
        dds.setTimeBetweenConnectErrorMillis(fsoConfig.getDuckdb().getV2().getTimeBetweenConnectErrorMillis());
        dds.setTestWhileIdle(true);
        dds.setValidationQuery(TEST_SQL);
        return dds;
    }

    /**
     * Test the connection
     */
    public boolean test(String tenantId, DataSource ds) {
        try (Connection conn = ds.getConnection(); Statement stmt = conn.createStatement()) {
            stmt.addBatch(BATCH_SQL_1);
            stmt.addBatch(BATCH_SQL_2);
            stmt.addBatch(BATCH_SQL_3);
            stmt.addBatch(BATCH_SQL_4);

//  When connecting, only attach once
//  duckLakeAttach  ->  ducklake:postgres:dbname=ducklake_meta user=postgres password=123456 host=localhost port=5432
//            String attachSQL = String.format("ATTACH '%s' as tenant_lake (DATA_PATH '%s')", fsoConfig.getDuckdb().getV2().getDuckLakeAttach(), fsoConfig.completeDuckdbV2DataPath(tenantId));
//            stmt.addBatch(attachSQL);
//            stmt.addBatch(BATCH_SQL_5);

            stmt.executeBatch();
            log.info("【DB】Connection test: Connection successful, url: {}", conn.getMetaData().getURL());
            return true;
        } catch (Exception e) {
            String error = String.format("【DB】Connection Test: Connection Failed, Cause by: %s", e);
            Assert.isFalse(false, error);
            return false;
        }
    }

    /**
     * Add a data source
     */
    public DataSource add(String tenantId, DataSource sds) {
        if (test(tenantId, sds)) {
            DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
            ds.addDataSource(getKey(tenantId), sds);
            return ds;
        }

        return null;
    }

    /**
     * Get the data source
     */
    public DataSource get(String tenantId) {
        DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
        return ds.getDataSource(getKey(tenantId));
    }

    /**
     * Whether the data source exists
     */
    public boolean contains(String tenantId) {
        DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
        return ds.getCurrentDataSources().containsKey(getKey(tenantId));
    }

    /**
     * Deleting a Data Source
     */
    public void remove(String tenantId) {
        DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
        ds.removeDataSource(getKey(tenantId));
    }

    /**
     * Initialize the data source
     */
    public synchronized void init(String tenantId) {
        if (contains(tenantId)) {
            return;
        }

        DataSource source = create();
        add(tenantId, source);
    }

    /**
     * Data source operations
     */
    public <T, R> R operator(String tenantId, T t, Function<T, R> function) {
        // Initialize the data source
        init(tenantId);

        // Supplementary data sources
        try {
            DynamicDataSourceContextHolder.push(getKey(tenantId));
            log.info("【DS】query, Param: {}", t);
            return function.apply(t);
        } catch (Exception e) {
            log.error("【DS】operator error, Cause by:", e);
            throw new AssertException(e.getMessage());
        } finally {
            DynamicDataSourceContextHolder.poll();
        }

    }

    /**
     * ds key
     */
    public String getKey(String tenantId) {
        return DUCKDB + tenantId;
    }

[staticlibs]

@DSir-ai

I suggest to use jdbc_pin_db and do the ATTACH only once on pool init. I think this may help here, because the whole problem sounds like losing an attached DB on in-memory "main" DB instance destroy (after the pool closes the last connection to it).

Alternatively it may be more convenient to use jdbc:duckdb:ducklake:<ducklake_url> as the "main" URL instead of connecting to jdbc:duckdb: and then calling USE tenant_lake.

[DSir-ai]

@staticlibs

I haven't seen any official way to connect directly to a URL.

They all just load the plugin and then attach it.

What is the specific format for connecting to postgres using the jdbc:duckdb:ducklake:<ducklake_url> method?

[staticlibs]

@DSir-ai

DuckLake supports the "direct attach" mode (ref) when the ducklake:... URL is used as the main URL instead of the ATTACH command.

With JDBC the DuckLake URL needs to be passes as part of the JDBC connection string, see details (and the URL example) in #258.

[DSir-ai]

@staticlibs

I have tried connecting to postgres using url and it works fine.

Can I specify data_path in url?

[staticlibs]

@DSir-ai

No, data path cannot be specified with direct attach - it can only be used when connecting to a pre-existing DuckLake instance.

[DSir-ai]

@staticlibs

Where is the data_path for the connection set?