add run.py and v0 of the app

Dockerfile

@@ -17,10 +17,10 @@ WORKDIR $HOME/app
 RUN pip install --no-cache-dir --upgrade pip
 
 # Copy the current directory contents into the container at $HOME/app setting the owner to the user
-COPY --chown=user run.py app.py requirements.txt $HOME/app
+COPY --chown=user run.py app.py requirements.txt $HOME/app/
 
 # Install dependencies
-RUN pip install gradio
+RUN pip install "gradio[oauth]" fire
 RUN pip install -r requirements.txt
 
 # Run app

README.md

@@ -1,10 +1,14 @@
 ---
 title: Run Duckdb
-emoji: 🐠
-colorFrom: indigo
-colorTo: purple
+emoji: 🦆
+colorFrom: yellow
+colorTo: green
 sdk: docker
 pinned: false
+hf_oauth: true
+hf_oauth_scopes:
+ - manage-repos
+ - jobs-api
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -1,20 +1,127 @@
+import re
 import subprocess
-import sys
+import yaml
 
 import gradio as gr
+import requests
+from huggingface_hub import HfApi
 
+CMD = ["python" ,"run.py"]
 
-def run():
-    process = subprocess.Popen(["python", "run.py"], stdout=subprocess.PIPE)
+with open("README.md") as f:
+    METADATA = yaml.safe_load(f.read().split("---\n")[1])
+TITLE = METADATA["title"]
+EMOJI = METADATA["emoji"]
+
+try:
+    process = subprocess.run(CMD + ["--help"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    HELP = not process.returncode and (process.stdout or process.stderr).decode()
+except Exception:
+    HELP = False
+
+DRY_RUN = bool(HELP) and bool(m :=re.search("--dry(-|_)run", HELP)) and m.group(0)
+
+def update_pbars(pbars: dict[str, float], line: str):
+    if (percent_match := re.search("\\d+(?:\\.\\d+)?%", line)) and any(c in line.split("%")[1][:10] for c in "|█▌"):
+        [pbars.pop(desc) for desc, percent in pbars.items() if percent == 1.]
+        percent = float(percent_match.group(0)[:-1]) / 100
+        desc = line[:percent_match.start()].strip() or "Progress"
+        pbars[desc] = percent
+
+def dry_run(src, config, split, dst, query):
+    if not all([src, config, split, dst, query]):
+        raise gr.Error("Please fill source, destination and query.")
+    process = subprocess.Popen(CMD + ["--src", src, "--config", config, "--split", split, "--dst", dst, "--query", query, DRY_RUN], stdout=subprocess.PIPE)
     logs = ""
     for line in iter(process.stdout.readline, b""):
         logs += line.decode()
-        yield logs
+        yield {output_markdown: logs, progress_labels: gr.Label(visible=False)}
+
+def run(src, config, split, dst, query):
+    if not all([src, config, split, dst, query]):
+        raise gr.Error("Please fill source, destination and query.")
+    raise gr.Error("NotImplemented")
+
+READ_FUNCTIONS = ("pl.read_parquet", "pl.read_csv", "pl.read_json")
+NUM_TRENDING_DATASETS = 10
 
 with gr.Blocks() as demo:
-    button = gr.Button("Run")
-    output_textbox = gr.Textbox()
-    button.click(run, outputs=[output_textbox])
+    with gr.Row():
+        with gr.Column(scale=10):
+            gr.Markdown(f"# {TITLE} {EMOJI}")
+        with gr.Column():
+            gr.LoginButton(scale=0.1)
+    with gr.Row():
+        with gr.Column():
+            with gr.Row():
+                loading_codes_json = gr.JSON([], visible=False)
+                dataset_dropdown = gr.Dropdown(label="Source Dataset", allow_custom_value=True, scale=10)
+                subset_dropdown = gr.Dropdown(info="Subset", allow_custom_value=True, show_label=False, visible=False)
+                split_dropdown = gr.Dropdown(info="Split", allow_custom_value=True, show_label=False, visible=False)
+        with gr.Column(scale=0.1, min_width=60):
+            gr.HTML("<div style='font-size: 4em;'>→</div>")
+        with gr.Column():
+            dst_dropdown = gr.Dropdown(label="Destination Dataset", allow_custom_value=True)
+    query_textarea = gr.TextArea(label="SQL Query", placeholder="SELECT * FROM src;", value="SELECT * FROM src;", container=False, show_label=False)
+    with gr.Row():
+        run_button = gr.Button("Run", scale=10, variant="primary")
+        if DRY_RUN:
+            dry_run_button = gr.Button("Dry-Run")
+    progress_labels= gr.Label(visible=False, label="Progress")
+    output_markdown = gr.Markdown(label="Output logs")
+    run_button.click(run, inputs=[dataset_dropdown, subset_dropdown, split_dropdown, dst_dropdown, query_textarea], outputs=[progress_labels, output_markdown])
+    if DRY_RUN:
+        dry_run_button.click(dry_run, inputs=[dataset_dropdown, subset_dropdown, split_dropdown, dst_dropdown, query_textarea], outputs=[progress_labels, output_markdown])
+
+    def show_subset_dropdown(dataset: str):
+        if dataset and "/" not in dataset.strip().strip("/"):
+            return []
+        resp = requests.get(f"https://datasets-server.huggingface.co/compatible-libraries?dataset={dataset}", timeout=3).json()
+        loading_codes = ([lib["loading_codes"] for lib in resp.get("libraries", []) if lib["function"] in READ_FUNCTIONS] or [[]])[0] or []
+        subsets = [loading_code["config_name"] for loading_code in loading_codes]
+        subset = (subsets or [""])[0]
+        return dict(choices=subsets, value=subset, visible=len(subsets) > 1, key=hash(str(loading_codes))), loading_codes
+
+    def show_split_dropdown(subset: str, loading_codes: list[dict]):
+        splits = ([list(loading_code["arguments"]["splits"]) for loading_code in loading_codes if loading_code["config_name"] == subset] or [[]])[0]
+        split = (splits or [""])[0]
+        return dict(choices=splits, value=split, visible=len(splits) > 1, key=hash(str(loading_codes) + subset))
+
+    @demo.load(outputs=[dataset_dropdown, loading_codes_json, subset_dropdown, split_dropdown])
+    def _fetch_datasets(request: gr.Request):
+        dataset = "CohereForAI/Global-MMLU"
+        datasets = [dataset] + [ds.id for ds in HfApi().list_datasets(limit=NUM_TRENDING_DATASETS, sort="trendingScore", direction=-1) if ds.id != dataset]
+        subsets, loading_codes = show_subset_dropdown(dataset)
+        splits = show_split_dropdown(subsets["value"], loading_codes)
+        return {
+            dataset_dropdown: gr.Dropdown(choices=datasets, value=dataset),
+            loading_codes_json: loading_codes,
+            subset_dropdown: gr.Dropdown(**subsets),
+            split_dropdown: gr.Dropdown(**splits),
+        }
+    
+    @dataset_dropdown.select(inputs=[dataset_dropdown], outputs=[subset_dropdown, split_dropdown])
+    def _show_subset_dropdown(dataset: str):
+        subsets, loading_codes = show_subset_dropdown(dataset)
+        splits = show_split_dropdown(subsets["value"], loading_codes)
+        return {
+            subset_dropdown: gr.Dropdown(**subsets),
+            split_dropdown: gr.Dropdown(**splits),
+        }
+    
+    @subset_dropdown.select(inputs=[dataset_dropdown, subset_dropdown, loading_codes_json], outputs=[split_dropdown])
+    def _show_split_dropdown(dataset: str, subset: str, loading_codes: list[dict]):
+        splits = show_split_dropdown(subset, loading_codes)
+        return {
+            split_dropdown: gr.Dropdown(**splits),
+        }
+
+if HELP:
+    with demo.route("Help", "/help"):
+        gr.Markdown(f"# Help\n\n```\n{HELP}\n```")
+
+with demo.route("Jobs", "/jobs"):
+    gr.Markdown("# Jobs")
 
 if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch(server_name="0.0.0.0")

requirements.txt

@@ -1,2 +1,3 @@
 duckdb
-huggingface_hub
+huggingface_hub
+tabulate

run.py

@@ -1,2 +1,66 @@
-print("hello world")
-print("done")
+import fire
+
+CONFIG = {
+    "preserve_insertion_order": False
+}
+
+CMD_SRC_KWARGS = """
+SELECT ('hf://datasets/{src}/' || lo.arguments['splits']['{split}']) AS path, function
+FROM (
+    SELECT unnest(li.loading_codes) AS lo, li.function[4:] as function
+    FROM (
+        SELECT unnest(libraries) as li
+        FROM read_json('https://datasets-server.huggingface.co/compatible-libraries?dataset={src}')
+    ) WHERE li.function[:3] = 'pl.'
+) WHERE lo.config_name='{config}';
+""".strip()
+
+CMD_SRC = """
+CREATE VIEW src AS SELECT * FROM {function}('{path}');
+""".strip()
+
+
+CMD_DST = """
+COPY ({query}) to 'tmp' (FORMAT PARQUET, ROW_GROUP_SIZE_BYTES '100MB', ROW_GROUPS_PER_FILE 5, PER_THREAD_OUTPUT true);
+""".strip()
+
+CMD_SRC_DRY_RUN = CMD_SRC[:-1] + " LIMIT 5;"
+CMD_DST_DRY_RUN = "{query};"
+
+def sql(src: str, dst: str, query: str, config: str = "default", split: str = "train", private: bool = False, dry_run: bool = False):
+    import os
+    import duckdb
+    from contextlib import nullcontext
+    from huggingface_hub import CommitScheduler
+
+    class CommitAndCleanScheduler(CommitScheduler):
+
+        def push_to_hub(self):
+            for path in self.folder_path.with_name("tmp").glob(self.allow_patterns):
+                with path.open("rb") as f:
+                    footer = f.read(4) and f.seek(-4, os.SEEK_END) and f.read(4)
+                if footer == b"PAR1":
+                    path.rename(self.folder_path / path.name)
+            super().push_to_hub()
+            for path in self.last_uploaded:
+                path.unlink(missing_ok=True)
+
+    with nullcontext() if dry_run else CommitAndCleanScheduler(repo_id=dst, repo_type="dataset", folder_path="dst", path_in_repo="data", allow_patterns="*.parquet", every=0.1, private=private):
+        con = duckdb.connect(":memory:", config=CONFIG)
+        src_kwargs = con.sql(CMD_SRC_KWARGS.format(src=src, config=config, split=split)).df().to_dict(orient="records")
+        if not src_kwargs:
+            raise ValueError(f'Invalid --config "{config}" for dataset "{src}", please select a valid dataset config/subset.')
+        con.sql((CMD_SRC_DRY_RUN if dry_run else CMD_SRC).format(**src_kwargs[0]))
+        if dry_run:
+            print(f"Sample data from '{src}' that would be written to '{dst}':\n")
+        else:
+            con.sql("PRAGMA enable_progress_bar;")
+        result = con.sql((CMD_DST_DRY_RUN if dry_run else CMD_DST).format(query=query.rstrip("\n ;")))
+        if dry_run:
+            print(result.df().to_markdown())
+        else:
+            print("done")
+    
+
+if __name__ == '__main__':
+    fire.Fire(sql)