Add upload csv functionality

.env.example

@@ -45,6 +45,9 @@ FDA_PG_POOL_IDLE_TIMEOUT_MS=10000
 FDA_PG_POOL_CONN_TIMEOUT_MS=5000
 FDA_PG_POOL_DB_IDLE_TIMEOUT_MS=300000
 
+## Upload files
+FDA_MAX_UPLOAD_SIZE=52428800
+
 ## Object Bucket-Based Storage System
 FDA_OBJSTG_USER=exampleUser
 FDA_OBJSTG_PASSWORD=examplePass

CHANGES_NEXT_RELEASE

@@ -0,0 +1 @@
+- Add: file upload endpoint POST /{visibility}/fdas/upload supporting CSV, XLS and XLSX files with configurable size limit (FDA_MAX_UPLOAD_SIZE) and automatic Parquet conversion (#301)

doc/05_advanced_topics.md

@@ -138,6 +138,41 @@ coordination, role-based scaling, query lifecycle behavior, fresh-query limits,
 
 ---
 
+## Uploading Tabular Data
+
+FDA supports publishing tabular data directly from CSV, XLS, or XLSX files via the `POST /{visibility}/fdas/upload`
+endpoint. This feature is ideal for static datasets that do not come from a live database.
+
+#### File Format Support
+
+-   **CSV**: Detected by MIME type `text/csv` or extension `.csv`. Parsed using `csv-parse` with automatic delimiter
+    detection.
+-   **XLS / XLSX**: Detected by MIME types `application/vnd.ms-excel` and
+    `application/vnd.openxmlformats-officedocument.spreadsheetml.sheet`, or extensions `.xls`/`.xlsx`. Data is read from
+    every sheet with a valid header row, headers are unioned, and rows are merged into one canonical CSV representation.
+
+#### Conversion Process
+
+1. The file is stored temporarily (either on disk or in object storage) to avoid keeping the entire buffer in memory or
+   in the job queue.
+2. The content is parsed and converted to a canonical CSV format.
+3. The CSV is uploaded to object storage under a temporary key (`tmp/<fdaId>_<timestamp>.csv`).
+4. DuckDB reads the CSV using `read_csv_auto` and exports it to Parquet with optional partitioning and compression.
+5. The temporary CSV is deleted.
+
+#### Handling Large Files
+
+The maximum file size is controlled by the environment variable `FDA_MAX_UPLOAD_SIZE` (default 50 MB). This limit can be
+adjusted to suit your infrastructure. For larger files, consider using direct S3 uploads or chunked streaming, but the
+current implementation is designed for moderate-sized datasets.
+
+#### Automatic DA Creation
+
+If the instance default or the request enables it, a `defaultDataAccess` DA is created after a successful upload. This
+DA provides basic querying capabilities (e.g., `SELECT * LIMIT 10`) and can be extended with custom parameters.
+
+---
+
 ## Pentaho CDA Compatibility Layer
 
 FDA includes a compatibility layer to support legacy Pentaho CDA clients.

doc/06_testing.md

@@ -75,6 +75,7 @@ Unit tests validate, among others:
 -   DA/FDA query composition
 -   error propagation and cleanup behavior
 -   API route wiring and request validation
+-   multipart upload endpoint validation (`/fdas/upload`) including file-type, size and form-field checks
 
 Integration tests validate:
 
@@ -84,6 +85,7 @@ Integration tests validate:
 -   End-to-end API behavior using real data
 -   direct FDA fresh execution over PostgreSQL
 -   default DA behavior including automatic creation and optional filters
+-   multipart upload flow (CSV/XLS/XLSX) from API ingestion to parquet generation and queryability
 
 ---
 
@@ -111,6 +113,7 @@ Current integration structure:
 -   `test/integration/suites/platform.integration.tests.js`: platform endpoints (`/health`, `/metrics`)
 -   `test/integration/suites/fdaCreation.integration.tests.js`: FDA creation and basic lifecycle creation checks
 -   `test/integration/suites/slidingWindows.integration.tests.js`: sliding-window and partitioning checks
+-   `test/integration/suites/uploadFdas.integration.tests.js`: multipart upload endpoint and upload-processing scenarios
 
 ---
 

doc/AdvancedTopics/sliding_windows_and_partitioning.md

@@ -14,13 +14,13 @@ reprocessing the entire dataset, only the most recent slice of data is fetched a
 Each configuration parameter is described in detail in the [API documentation](../03_api.md/#refresh-policy-object), but
 the key fields are summarized below:
 
-| Field                        | Description                                                                                                                               |
-| ---------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- |
-| `type`                       | Must be set to `window` to enable sliding window behavior.                                                                                |
-| `refreshInterval`            | Defines how often the window refresh runs. It accepts either a human interval such as `1 hour` or a cron expression.                      |
-| `consistencyRefreshInterval` | Optional lower-frequency schedule for a full FDA rebuild that helps recover delayed historical data. It must be greater than `refreshInterval`, otherwise `InvalidParam` error response will be got                                      |
-| `fetchSize`                  | Defines the **time range** of data to fetch on each refresh. For example, `week` fetches data from the last week.                         |
-| `windowSize`                 | Specifies the total retention window (e.g., data from last month), defining which data should be preserved and which should be discarded. |
+| Field                        | Description                                                                                                                                                                                         |
+| ---------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `type`                       | Must be set to `window` to enable sliding window behavior.                                                                                                                                          |
+| `refreshInterval`            | Defines how often the window refresh runs. It accepts either a human interval such as `1 hour` or a cron expression.                                                                                |
+| `consistencyRefreshInterval` | Optional lower-frequency schedule for a full FDA rebuild that helps recover delayed historical data. It must be greater than `refreshInterval`, otherwise `InvalidParam` error response will be got |
+| `fetchSize`                  | Defines the **time range** of data to fetch on each refresh. For example, `week` fetches data from the last week.                                                                                   |
+| `windowSize`                 | Specifies the total retention window (e.g., data from last month), defining which data should be preserved and which should be discarded.                                                           |
 
 ---
 

doc/postman/README.md

@@ -51,8 +51,39 @@ For `GET /{visibility}/fdas/{fdaId}/das/{daId}/data`, response format is selecte
 The same content negotiation rules apply to `GET /{visibility}/fdas/{fdaId}/data`, which is the direct fresh FDA query
 endpoint.
 
+---
+
 ### 4. Authentication
 
 FIWARE-data-access does not implement any specific security mechanisms, it offers the `visibility` property so the user
 can build his auth system on top. For those cases we added an optional `auth_token` variable to the requests thats gonna
 be mapped to the `X-Auth-Token` header, in case the user has a system that requires token-based auth.
+
+---
+
+### 5. Uploading CSV/XLS/XLSX with Postman
+
+To create FDAs from tabular files use `POST /{{visibility}}/fdas/upload` with **Body = form-data**:
+
+-   Required fields:
+    -   `id` (Text)
+    -   `file` (File)
+-   Optional fields:
+    -   `description` (Text)
+    -   `timeColumn` (Text)
+    -   `objStgConf` (Text, JSON object string)
+    -   `defaultDataAccess` (Text/Boolean)
+    -   `datasourceId` (Text)
+
+Example `objStgConf` value:
+
+```json
+{ "partition": "day", "compression": "zstd" }
+```
+
+Expected behavior:
+
+-   `202` when upload is accepted and processing continues asynchronously.
+-   `400` for invalid upload fields (`objStgConf.partition`, `objStgConf.compression`, malformed JSON, missing file).
+-   `413` when file size exceeds the configured limit.
+-   `415` when file type is not CSV/XLS/XLSX.

package.json

@@ -32,17 +32,20 @@
     "@duckdb/node-api": "~1.5.2-r.1",
     "agenda": "~6.2.5",
     "ajv": "~8.17.1",
+    "cron-parser": "~5.5.0",
     "dotenv": "~17.2.3",
     "exceljs": "~4.4.0",
     "express": "5.1.0",
     "lodash": "~4.17.21",
     "logops": "~2.1.2",
     "mongodb": "~7.0.0",
+    "multer": "~1.4.5-lts.1",
     "pg": "~8.16.3",
     "pg-copy-streams": "~7.0.0",
     "pg-cursor": "~2.19.0",
     "uuid": "~11.1.0",
-    "cron-parser": "~5.5.0"
+    "xlsx": "~0.18.5",
+    "csv-parse": "~7.0.0"
   },
   "devDependencies": {
     "c8": "~10.1.3",

src/fetcher.js

@@ -23,7 +23,11 @@
 // criminal actions it may exercise to protect its rights.
 
 import { getAgenda } from './lib/jobs.js';
-import { cleanPartition, processFDAAsync } from './lib/fda.js';
+import {
+  cleanPartition,
+  processFDAAsync,
+  processUploadFDAJob,
+} from './lib/fda.js';
 import { getBasicLogger } from './lib/utils/logger.js';
 
 const logger = getBasicLogger();
@@ -75,6 +79,44 @@ export async function startFetcher() {
   agenda.define('clean-partition', cleanPartitionFDA);
   agenda.define('clean-partition-recurring', cleanPartitionFDA);
 
+  agenda.define('upload-fda', async (job) => {
+    const {
+      fdaId,
+      service,
+      servicePath,
+      visibility,
+      tempFilePath,
+      originalname,
+      mimetype,
+      description,
+      timeColumn,
+      objStgConf,
+      cached,
+      defaultDataAccessEnabled,
+      datasourceId,
+    } = job.attrs.data;
+
+    try {
+      await processUploadFDAJob({
+        fdaId,
+        service,
+        servicePath,
+        visibility,
+        tempFilePath,
+        originalname,
+        mimetype,
+        description,
+        timeColumn,
+        objStgConf,
+        cached,
+        defaultDataAccessEnabled,
+        datasourceId,
+      });
+    } catch (e) {
+      logger.error('Fetcher error: ', e);
+    }
+  });
+
   await agenda.start();
   logger.info('[Fetcher] Agenda started');
 }