Function test_array_with_nulls

tests/python/test_array_binary_debug.py:80–168

Test array with NULL values using psycopg3 binary protocol.

()

    return result

def test_array_with_nulls():
    """Test array with NULL values using psycopg3 binary protocol."""

    print("=" * 80)
    print("Testing PostgreSQL array binary encoding with NULLs")
    print("=" * 80)

    # Connect to pgsqlite server on port 5433
    conn_str = "host=localhost port=5433 user=postgres dbname=test"

    with psycopg.connect(conn_str) as conn:
        with conn.cursor() as cur:
            # Create test table
            print("\n1. Creating test table...")
            cur.execute("""
                CREATE TABLE IF NOT EXISTS array_test (
                    id INTEGER PRIMARY KEY,
                    int_array INTEGER[]
                )
            """)

            # Insert test data - array with NULLs
            print("\n2. Inserting array with NULLs: [1, NULL, 3]")
            cur.execute("""
                INSERT INTO array_test (id, int_array)
                VALUES (1, ARRAY[1, NULL, 3]::INTEGER[])
                ON CONFLICT (id) DO UPDATE SET int_array = excluded.int_array
            """)

            # Query with binary result format
            print("\n3. Querying with binary protocol...")
            cur.execute("""
                SELECT int_array FROM array_test WHERE id = 1
            """, binary=True)

            row = cur.fetchone()

            if row and row[0]:
                # Get the raw binary data
                raw_data = row[0]

                print(f"\n4. Received {len(raw_data)} bytes of binary data")

                # Decode and analyze the binary format
                decoded = decode_binary_array(raw_data)

                print("\n5. Decoded array structure:")
                print(f"   - ndim: {decoded.get('ndim')}")
                print(f"   - dataoffset: {decoded.get('dataoffset')} (0x{decoded.get('dataoffset'):02x})")
                print(f"   - elemtype: {decoded.get('elemtype')} (OID for INT4)")

                if decoded.get('dimensions'):
                    print(f"   - dimensions: {decoded['dimensions']}")

                if decoded.get('null_bitmap'):
                    print(f"   - null_bitmap (hex): {decoded['null_bitmap']}")
                    print(f"   - null_bitmap (binary): {decoded['null_bitmap_binary']}")