Getting error message when running in lambda: message: "Too many open files"

[rob3000]

Hi' i'm currently running the deltra-rs rust library in lambda to do a merge query into delta lake. However i noticed i'm getting the following error:

Error: hyper::Error(Connect, ConnectError("tcp open error", Os { code: 24, kind: Uncategorized, message: "Too many open files" }))

It looks like its coming from the DynamoDBLock;

{
    "timestamp": "Mar 28 09:06:04.847",
    "level": "ERROR",
    "fields": {
        "message": "dynamodb client failed to write log entry: GenericDynamoDb { source: HttpDispatch(HttpDispatchError { message: \"Error during dispatch: error trying to connect: dns error: Too many open files (os error 24)\" }) }"
    },
    "target": "deltalake_aws::logstore"
}

I suspect the logstore is trying to create new connections when running through a batch for records to save to deltalake, I'm still quite new to rust, but when i normally talk to dynamoDB i create a a reference outside of the handler and pass the reference through, do i need to look at doing the same with the logstore? And if so how?

[rtyler]

👋 are you manually creating the log store or just opening the table and grabbing a reference off the table to its log store? I can imagine this happening if there's a recursion of some form or a loop where new connections are being created in a loop before older ones are expired, but I wouldn't expect this in normal operation and haven't seen it myself
🤔

[rob3000]

👋I'm just opening the table then running a merge operation on some incoming data from a queue. Its say is a loop for each batch of messages. Is there a way to close the connection after write/merge?

[rob3000]

Thought i'd include how i'm running the code in case there is something that stood out that i'm doing something silly.

async fn handle_transaction_merge(data: RecordBatch) -> MergeResult<DataFrame> {
    let ctx = SessionContext::new();

    let _ = ctx.register_batch("t", data);
    let maybe_df = ctx.table("t").await;

    let df = match maybe_df {
        Ok(df) => df,
        Err(err) => panic!("{:?}", err),
    };
    Ok(df)
}

async fn function_handler(table: &DeltaTable, ops: &DeltaOps, events: SqsEvent) -> Result<(), E> {
    let data = event.records.into();
    let batch = convert_to_batch(&table, &data);

    let df = handle_transaction_merge(batch).await.unwrap();
    let ops = DeltaOps(table.to_owned());

    let (table, metrics) = ops
        .merge(df, col("target.id").eq(col("source.id")))
        .with_source_alias("source")
        .with_target_alias("target")
        .with_safe_cast(true)
        .when_matched_update(|update| {
            update
                .update("description", col("source.description"))
        })?
        .when_not_matched_insert(|insert| {
            insert
                .set("id", col("source.id"))
                .set("description", col("source.description"))
        })?
        .await?;

    Ok(())
}

#[instrument]
async fn get_delta_store() -> (DeltaTable, DeltaOps) {
    let table_path = "s3://some-bucket";

    deltalake::aws::register_handlers(None);
    let maybe_table = deltalake::open_table(&table_path).await;

    let table = match maybe_table {
        Ok(table) => table,
        Err(DeltaTableError::NotATable(_)) => {
            println!("It doesn't look like our delta table has been created");
            crete_table(&table_path).await
        }
        Err(err) => panic!("oh no! {:?}", err),
    };
    let ops = DeltaOps(table.to_owned());
    return (table, ops);
}

#[tokio::main]
async fn main() -> Result<(), E> {
    // Initialize logger
    setup_tracing();

    let store = get_delta_store().await;
    let table = store.0;
    let ops = store.1;

    lambda_runtime::run(service_fn(|event: LambdaEvent<SqsEvent>| {
        let (event, ctx) = event.into_parts();
        function_handler(&table, &ops, event)
    }))
    .await?;
    Ok(())
}