Draft: Fix inserts in through tables

.env.sample

@@ -21,6 +21,9 @@
 # USE_ASYNC=False
 # JOIN_QUERIES=True
 # STREAM_RESULTS=True
+# db polling interval
+# POLL_INTERVAL=0.1
+# FILTER_CHUNK_SIZE=5000
 
 # Elasticsearch
 # ELASTICSEARCH_SCHEME=http

.github/workflows/python-build.yml

@@ -8,7 +8,7 @@ jobs:
     strategy:
       max-parallel: 4
       matrix:
-        python-version: [3.7, 3.8, 3.9, '3.10']
+        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
     services:
       postgres:
         image: debezium/postgres:15
@@ -25,7 +25,7 @@ jobs:
         ports:
           - 6379:6379
       elasticsearch:
-        image: docker.elastic.co/elasticsearch/elasticsearch:7.17.6
+        image: docker.elastic.co/elasticsearch/elasticsearch:7.17.7
         ports:
           - 9200:9200
           - 9300:9300
@@ -34,9 +34,9 @@ jobs:
           network.host: 127.0.0.1
           http.host: 0.0.0.0
     steps:
-    - uses: actions/checkout@v1
+    - uses: actions/checkout@v3
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v1
+      uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies

bin/bootstrap

@@ -6,7 +6,7 @@ import logging
 import click
 
 from pgsync.sync import Sync
-from pgsync.utils import get_config, load_config, show_settings
+from pgsync.utils import config_loader, get_config, show_settings
 
 logger = logging.getLogger(__name__)
 
@@ -54,7 +54,7 @@ def main(teardown, config, user, password, host, port, verbose):
 
     show_settings(config)
 
-    for document in load_config(config):
+    for document in config_loader(config):
         sync: Sync = Sync(
             document, verbose=verbose, repl_slots=False, **kwargs
         )

bin/es_mapping

@@ -8,7 +8,7 @@ from elasticsearch import Elasticsearch, helpers
 
 from pgsync.settings import ELASTICSEARCH_TIMEOUT, ELASTICSEARCH_VERIFY_CERTS
 from pgsync.urls import get_elasticsearch_url
-from pgsync.utils import get_config, load_config, timeit
+from pgsync.utils import config_loader, get_config, timeit
 
 logger = logging.getLogger(__name__)
 
@@ -114,7 +114,9 @@ def main(config):
     """Create custom NGram analyzer for the default mapping."""
 
     config: str = get_config(config)
-    for index in set([document["index"] for document in load_config(config)]):
+    for index in set(
+        [document["index"] for document in config_loader(config)]
+    ):
         create_es_mapping(index)
 
 

bin/parallel_sync

@@ -44,6 +44,7 @@ and row numbers.
 import asyncio
 import multiprocessing
 import os
+import re
 import sys
 from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
 from dataclasses import dataclass
@@ -54,9 +55,32 @@ from typing import Generator, Optional, Union
 import click
 import sqlalchemy as sa
 
-from pgsync.settings import BLOCK_SIZE
+from pgsync.settings import BLOCK_SIZE, CHECKPOINT_PATH
 from pgsync.sync import Sync
-from pgsync.utils import get_config, load_config, show_settings, timeit
+from pgsync.utils import (
+    compiled_query,
+    config_loader,
+    get_config,
+    show_settings,
+    timeit,
+)
+
+
+def save_ctid(page: int, row: int, name: str) -> None:
+    checkpoint_file: str = os.path.join(CHECKPOINT_PATH, f".{name}.ctid")
+    with open(checkpoint_file, "w+") as fp:
+        fp.write(f"{page},{row}\n")
+
+
+def read_ctid(name: str) -> None:
+    checkpoint_file: str = os.path.join(CHECKPOINT_PATH, f".{name}.ctid")
+    if os.path.exists(checkpoint_file):
+        with open(checkpoint_file, "r") as fp:
+            pairs: str = fp.read().split()[0].split(",")
+            page = int(pairs[0])
+            row = int(pairs[1])
+            return page, row
+    return None, None
 
 
 def logical_slot_changes(
@@ -92,17 +116,71 @@ class Task:
 
 
 @timeit
-def fetch_tasks(doc: dict, block_size: Optional[int] = None) -> Generator:
+def fetch_tasks(
+    doc: dict,
+    block_size: Optional[int] = None,
+) -> Generator:
     block_size = block_size or BLOCK_SIZE
     pages: dict = {}
     sync: Sync = Sync(doc)
+    page: Optional[int] = None
+    row: Optional[int] = None
+    name: str = re.sub(
+        "[^0-9a-zA-Z_]+", "", f"{sync.database.lower()}_{sync.index}"
+    )
+    page, row = read_ctid(name=name)
     statement: sa.sql.Select = sa.select(
         [
             sa.literal_column("1").label("x"),
             sa.literal_column("1").label("y"),
             sa.column("ctid"),
         ]
     ).select_from(sync.tree.root.model)
+
+    # filter by Page
+    if page:
+        statement = statement.where(
+            sa.cast(
+                sa.func.SPLIT_PART(
+                    sa.func.REPLACE(
+                        sa.func.REPLACE(
+                            sa.cast(sa.column("ctid"), sa.Text),
+                            "(",
+                            "",
+                        ),
+                        ")",
+                        "",
+                    ),
+                    ",",
+                    1,
+                ),
+                sa.Integer,
+            )
+            > page
+        )
+
+    # filter by Row
+    if row:
+        statement = statement.where(
+            sa.cast(
+                sa.func.SPLIT_PART(
+                    sa.func.REPLACE(
+                        sa.func.REPLACE(
+                            sa.cast(sa.column("ctid"), sa.Text),
+                            "(",
+                            "",
+                        ),
+                        ")",
+                        "",
+                    ),
+                    ",",
+                    2,
+                ),
+                sa.Integer,
+            )
+            > row
+        )
+
     i: int = 1
     for _, _, ctid in sync.fetchmany(statement):
         value: list = ctid[0].split(",")
@@ -175,7 +253,6 @@ def multithreaded(
         queue.put(task)
 
     queue.join()  # block until all tasks are done
-
     logical_slot_changes(doc, verbose=verbose, validate=validate)
 
 
@@ -194,7 +271,6 @@ def multiprocess(
             list(executor.map(task.process, tasks))
         except Exception as e:
             sys.stdout.write(f"Exception: {e}\n")
-
     logical_slot_changes(doc, verbose=verbose, validate=validate)
 
 
@@ -209,13 +285,9 @@ def multithreaded_async(
     sys.stdout.write("Multi-threaded async\n")
     executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=nprocs)
     event_loop = asyncio.get_event_loop()
-    try:
-        event_loop.run_until_complete(
-            run_tasks(executor, tasks, doc, verbose=verbose, validate=validate)
-        )
-    finally:
-        event_loop.close()
-
+    event_loop.run_until_complete(
+        run_tasks(executor, tasks, doc, verbose=verbose, validate=validate)
+    )
     logical_slot_changes(doc, verbose=verbose, validate=validate)
 
 
@@ -234,9 +306,8 @@ def multiprocess_async(
         event_loop.run_until_complete(
             run_tasks(executor, tasks, doc, verbose=verbose, validate=validate)
         )
-    finally:
-        event_loop.close()
-
+    except KeyboardInterrupt:
+        pass
     logical_slot_changes(doc, verbose=verbose, validate=validate)
 
 
@@ -247,24 +318,19 @@ async def run_tasks(
     verbose: bool = False,
     validate: bool = False,
 ) -> None:
-    event_loop = asyncio.get_event_loop()
+    sync: Optional[Sync] = None
     if isinstance(executor, ThreadPoolExecutor):
         # threads can share a common Sync object
-        sync: Sync = Sync(doc, verbose=verbose, validate=validate)
-        tasks: list = [
-            event_loop.run_in_executor(
-                executor, run_task, task, sync, None, verbose, validate
-            )
-            for task in tasks
-        ]
-    else:
-        tasks = [
+        sync = Sync(doc, verbose=verbose, validate=validate)
+    event_loop = asyncio.get_event_loop()
+    completed, pending = await asyncio.wait(
+        [
             event_loop.run_in_executor(
-                executor, run_task, task, None, doc, verbose, validate
+                executor, run_task, task, sync, doc, verbose, validate
             )
             for task in tasks
         ]
-    completed, pending = await asyncio.wait(tasks)
+    )
     results: list = [task.result() for task in completed]
     print("results: {!r}".format(results))
     print("exiting")
@@ -286,7 +352,14 @@ def run_task(
         sync.index,
         sync.sync(ctid=task, txmin=txmin, txmax=txmax),
     )
-    print("run_task complete")
+    if len(task) > 0:
+        page: int = max(task.keys())
+        row: int = max(task[page])
+        name: str = re.sub(
+            "[^0-9a-zA-Z_]+", "", f"{sync.database.lower()}_{sync.index}"
+        )
+        save_ctid(page=page, row=row, name=name)
+
     return 1
 
 
@@ -331,14 +404,13 @@ def main(config, nprocs, mode, verbose):
     """
     TODO:
     - Track progress across cpus/threads
-    - Save ctid
     - Handle KeyboardInterrupt Exception
     """
 
     show_settings()
     config: str = get_config(config)
 
-    for document in load_config(config):
+    for document in config_loader(config):
         tasks: Generator = fetch_tasks(document)
         if mode == "synchronous":
             synchronous(tasks, document, verbose=verbose)

docker-compose.yml

@@ -13,7 +13,7 @@ services:
     image: redis
     command: redis-server --requirepass PLEASE_CHANGE_ME
   elasticsearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.6
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.7
     ports:
       - "9201:9200"
       - "9301:9300"

examples/airbnb/data.py

@@ -7,7 +7,7 @@
 
 from pgsync.base import pg_engine, subtransactions
 from pgsync.helper import teardown
-from pgsync.utils import get_config, load_config
+from pgsync.utils import config_loader, get_config
 
 
 @click.command()
@@ -21,7 +21,7 @@ def main(config):
 
     config: str = get_config(config)
     teardown(drop_db=False, config=config)
-    document: dict = next(load_config(config))
+    document: dict = next(config_loader(config))
     database: str = document.get("database", document["index"])
     with pg_engine(database) as engine:
         Session = sessionmaker(bind=engine, autoflush=True)

examples/airbnb/schema.py

@@ -7,7 +7,7 @@
 
 from pgsync.base import create_database, pg_engine
 from pgsync.helper import teardown
-from pgsync.utils import get_config, load_config
+from pgsync.utils import config_loader, get_config
 
 Base = declarative_base()
 
@@ -91,7 +91,7 @@ class Review(Base):
 
 
 def setup(config: str) -> None:
-    for document in load_config(config):
+    for document in config_loader(config):
         database: str = document.get("database", document["index"])
         create_database(database)
         with pg_engine(database) as engine:

examples/ancestry/data.py

@@ -4,7 +4,7 @@
 
 from pgsync.base import pg_engine, subtransactions
 from pgsync.helper import teardown
-from pgsync.utils import get_config, load_config
+from pgsync.utils import config_loader, get_config
 
 
 @click.command()
@@ -18,7 +18,7 @@ def main(config):
 
     config: str = get_config(config)
     teardown(drop_db=False, config=config)
-    document: dict = next(load_config(config))
+    document: dict = next(config_loader(config))
     database: str = document.get("database", document["index"])
     with pg_engine(database) as engine:
         Session = sessionmaker(bind=engine, autoflush=True)

examples/ancestry/schema.py

@@ -4,7 +4,7 @@
 
 from pgsync.base import create_database, pg_engine
 from pgsync.helper import teardown
-from pgsync.utils import get_config, load_config
+from pgsync.utils import config_loader, get_config
 
 Base = declarative_base()
 
@@ -49,7 +49,7 @@ class GreatGrandChild(Base):
 
 
 def setup(config: str) -> None:
-    for document in load_config(config):
+    for document in config_loader(config):
         database: str = document.get("database", document["index"])
         create_database(database)
         with pg_engine(database) as engine: