Oracle upsert perfomance issue to read the bulk data_insert using merge statement.

[Jeevananthan-23]

async def fast_oracle_bulk_merge(data_insert):
    """
    High-performance Oracle bulk MERGE operation
    Returns number of processed rows
    """
    if not data_insert:
        return 0

    try:
        # 1. Establish connection with optimized settings
        conn = await oracledb.connect_async(
            user=DB_USER,
            password=DB_PASSWORD,
            dsn=f"{DB_HOST}:{DB_PORT}/{DB_SERVICENAME}",
            threaded=True,  # Better for bulk operations
            events=False,   # Disable unused features
            cclass="BULK_MERGE"  # Connection class for tracking
        )

        # 2. Prepare MERGE statement with numbered parameters
        merge_sql = """
        MERGE INTO TB_ABSA_TRAIN_DATA_MASTER t
        USING (SELECT :1 AS category_name, :2 AS keyword FROM dual) src
        ON (t.CATEGORY_NAME = src.category_name AND t.KEYWORD = src.keyword)
        WHEN NOT MATCHED THEN
            INSERT (CATEGORY_NAME, KEYWORD, CATEGORY_CODE, SOURCE, STATUS)
            VALUES (:1, :2, :3, :4, :5)
        """

        # 3. Process in batches with optimal settings
        batch_size = 10000  # Adjust based on testing
        processed = 0
        
        for i in range(0, len(data_insert), batch_size):
            batch = data_insert[i:i + batch_size]
            
            # Convert to tuples for better performance
            params = [
                (
                    item['category_name'],
                    item['keyword'],
                    item['category_code'],
                    item['source'],
                    item['status']
                )
                for item in batch
            ]

            # 4. Execute with optimized settings
            cursor = conn.cursor()
            await cursor.executemany(
                merge_sql,
                params,
                batcherrors=True,
                batch_errors=True,
                arraydmlrowcounts=True
            )
            
            # 5. Get actual affected rows
            processed += sum(await cursor.getarraydmlrowcounts())
            
            # 6. Handle any batch errors
            for error in await cursor.getbatcherrors():
                logger.error(f"Row {error.offset} failed: {error.message}")
            
            await cursor.close()

        await conn.commit()
        return processed

    except Exception as e:
        logger.error(f"Bulk merge failed: {str(e)}")
        if 'conn' in locals():
            await conn.rollback()
        raise
    finally:
        if 'conn' in locals():
            await conn.close()

Very slow in performance my JS version is execute the same data within 11mins. Record count is ~100k.

[cjbj]

Which bit of this is slow? The Python language? The connection creation? The conversion in Python to tuples? The actual executemany() call? How do each of those compare with node-oracledb (I assume you mean this)?

Refer to https://python-oracledb.readthedocs.io/en/latest/user_guide/batch_statement.html#predefining-memory-areas, which can help avoid memory reallocations if your data has varying lengths.

[cjbj]

On the python-oracledb side you can control the

• creation of the tuples, though you don't have much choice about this, and your results indicate this isn't your biggest issue.
• memory allocation & type handling optimization via setinputsizes()
• the batch size aka number of records you pass to executemany()
• SDU size. Note in Thick mode the SDU is set in a sqlnet.ora file. In both modes you will probably need the DBA to bump the server-side SDU since there is a negotiation down between client & server settings. There are some links about tuning the SDU in the doc.

After that you should look at network performance and SQL statement optimization, see Round-trips are only part of the data insertion problem for a case study, also with MERGE. There are experts in other forums that can help you with these, but let us know how you go.

[anthony-tuininga]

Sending the data to the database (executemany): This is the slow part. It takes about 47-48 seconds for each group of 10,000 rows.

This is strange. I can insert 2.5 million (fairly small) rows in just over a second on my machine. 10,000 rows would take a handful of milliseconds. It would probably be good to figure out where the 47-48 seconds of time is happening. If you set the environment variable PYO_DEBUG_PACKETS to any value and run your script (making sure you dump output to a file) you can examine the timing involved -- in particular the time between the last send of data to the database and response from the database. I am going to assume that the majority of the time is waiting for the database to complete its work. And if my assumption is correct, the solution will have nothing to do with Python or the driver and everything to do with your database and its configuration or table structure.

[cjbj]

One quick datapoint would be to create a new table (with no indexes etc) and change your SQL statement to be a simple INSERT into that table.

[Mallikarjun19]

I've implemented a different approach, instead of using a MERGE statement, my new process is to first filter the data within the Python code to find only the new records, and then I perform a single bulk INSERT. The entire operation now completes in just 7-8 seconds,
is this approach good or should i stick to my old approach and try improving it?

async def import_category_xlsx(file_path: str):
    start_time = time.time()
    try:
        async with db.get_async_session() as session:
            logger.info(f"Processing from: {file_path}")
            # Read all sheets into a dict of DataFrames
            xls = pd.ExcelFile(file_path, engine=None)
    
            for sheet in xls.sheet_names:
                df = pd.read_excel(xls, sheet_name=sheet)
                valid_cells = df.stack().dropna().unique()

                # Use add_keywords to insert keywords for this sheet
                await add_keywords(sheet, valid_cells.tolist())

            elapsed = time.time() - start_time
            print("Elapsed Time: ",elapsed)
            
    except Exception as e:
        logger.error(f"Error processing {file_path}: {e}")
        try:
            await session.rollback()
        except Exception:
            pass
        raise e



async def check_existing_category(status, sess: AsyncSession, processed_category):
    query_cat = select(CategoryMaster).filter(CategoryMaster.category_name == processed_category, CategoryMaster.status == status, CategoryMaster.category_type == 'B')
    result_cat = await sess.execute(query_cat)
    category_obj = result_cat.scalars().all()
    return category_obj

async def add_keywords(category: str, keywords: List[str]):

    try:
        async with db.get_async_session() as session:
            category_name = category.strip().lower()
            code = category_name.replace(" ", "_")
            if not category_name or not keywords:
                raise ValueError("Category name and keywords cannot be empty")

            category_inst = await check_existing_category("ACTIVE", session, category_name)


            if category_inst:
                # Fetch existing keywords
                query_existing = select(TrainDataMaster.KEYWORD).filter_by(CATEGORY_NAME=category_name)
                result_existing = await session.execute(query_existing)
                existing_keywords = set((row[0]) for row in result_existing.all())

                # Process and filter new keywords
                keywords_to_insert = []
                for kw in keywords:
                    keyword = str(kw).strip().lower()
                    if keyword not in existing_keywords:
                        keywords_to_insert.append({
                            "CATEGORY_NAME": category_name,
                            "CATEGORY_CODE": code,
                            "KEYWORD": keyword,
                            "SOURCE": "API",
                            "STATUS": "ACTIVE"
                        })

                # Bulk insert
                if keywords_to_insert:
                    objs = [TrainDataMaster(**item) for item in keywords_to_insert]
                    session.add_all(objs)
                    return {
                        "message": "Keywords added successfully"
                    }
                else:
                    return {
                        "message": "No new keywords to add"
                    }
            else:
                raise ValueError("Category does not exist!")

    except Exception as e:
        logger.error(f"Error adding keywords: {e}")
        # Rollback only if session is active
        try:
            await session.rollback()
        except Exception:
            pass
        raise e

[SanoopAnanth]

Is it good practice to use synchronous read operation for excel file [pd.read_excel( ), pd.ExcelFile( )] in an async method? Instead use asyncio to thread it.