Best approach to materialize a QBV using Flux

[grtjn]

We have certain processes that effectively materialize QBV queries. It was suggested to us that it might work better if you use Flux to coordinate this. I was looking into this, but not 100% sure what the best approach would be. I'd prefer not doing a separate export and import, but copy isn't adequate, and worried that reprocess would not be optimal either.

Any thoughts?

[rjrudin]

My understanding is that custom code is needed to generate a QBV. ml-gradle has support for it - https://github.com/marklogic/ml-gradle/wiki/Loading-schemas#generating-query-based-views - and that relies on a call to v1/eval to generate the QBV, and ml-gradle then writes it to a schemas database. You could in theory use reprocess to do this, but how many QBVs are being generated? And do they need to be generated at runtime, or can they be part of the deployment step for an application?

[grtjn]

I think you misunderstood. We want to materialize the sql/optic result. We have already covered the generate part. :)

I think import from JDBC, and doing a blunt SELECT * could work. It would require having an ODBC connection available, but we have those for BI tools anyhow. Tapping into /v1/rows doesn't seem like an option? Not sure if that would scale well to be honest.

[rjrudin]

I definitely don't know what's going on here. :) What's the overall flow here? When is the QBV generated? Where is data being imported from, and/or where does data need to be exported to?

A QBV can be referenced by op.fromView, so any Flux export command that supports an op.fromView Optic query should work when using either a QBV or a view defined by a TDE.

[grtjn]

Source and target hosts are same for us. Most typical case is running the QBV in Staging, and materialize the output in Final. But it could be same database as well. Some QBV's produce a lot or records (several mln), and can contain complex joins or aggregates. I am reluctant writing my own naive optic code to read the QBV.

We have a mechanism in place, using Corb, but it is suboptimal.

[rjrudin]

Interesting - so you use a QBV in staging to pull out data as rows. And then you want to write those rows as JSON or XML documents to the final database?

So it's like a different type of copy. Copy of course queries for URIs and then copies documents to another database, which can use a REST transform.

Right now, only export commands support retrieving data as rows via Optic. Possibly export-marklogic as a command - i.e. export rows via an Optic query, and then serialize the rows to JSON or XML documents and write them via an "output connection", similar to copy?

[grtjn]

I am actually running a test with import-jdbc, which should allow me to 'grab' a full table, and also ingest it immediately. I instantly run into an exception that I cannot really explain though:

2025-11-26 19:41:07 com.marklogic.flux: Will connect to MarkLogic host: xxx
2025-11-26 19:41:08 org.postgresql.core.v3.QueryExecutorImpl: Suppress sending BEGIN from execute sendQueryPreamble.
2025-11-26 19:41:08 org.postgresql.core.v3.QueryExecutorImpl: Suppress sending BEGIN from execute sendQueryPreamble.
2025-11-26 19:41:08 org.postgresql.core.v3.QueryExecutorImpl: Suppress sending BEGIN from execute sendQueryPreamble.
2025-11-26 19:41:08 com.marklogic.flux: Displaying stacktrace due to use of --stacktrace option
com.marklogic.flux.api.FluxException: requirement failed: The length of varchar type cannot be negative.
	at com.marklogic.flux.impl.AbstractCommand.handleException(AbstractCommand.java:167) ~[flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.impl.AbstractCommand.execute(AbstractCommand.java:82) ~[flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.cli.Main.executeCommand(Main.java:138) ~[flux-cli-1.4.0.jar:?]
	at picocli.CommandLine.execute(CommandLine.java:2174) [picocli-4.7.6.jar:4.7.6]
	at com.marklogic.flux.cli.Main.run(Main.java:114) [flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.cli.Main.run(Main.java:79) [flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.cli.Main.main(Main.java:75) [flux-cli-1.4.0.jar:?]
Caused by: java.lang.IllegalArgumentException: requirement failed: The length of varchar type cannot be negative.
	at scala.Predef$.require(Predef.scala:281) ~[scala-library-2.12.18.jar:?]
	at org.apache.spark.sql.types.VarcharType.<init>(VarcharType.scala:23) ~[spark-sql-api_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.getCatalystType(JdbcUtils.scala:219) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$getSchema$1(JdbcUtils.scala:308) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at scala.Option.getOrElse(Option.scala:189) ~[scala-library-2.12.18.jar:?]
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.getSchema(JdbcUtils.scala:308) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.getQueryOutputSchema(JDBCRDD.scala:71) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:58) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation$.getSchema(JDBCRelation.scala:241) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:37) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:346) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:229) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.DataFrameReader.$anonfun$load$2(DataFrameReader.scala:211) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at scala.Option.getOrElse(Option.scala:189) ~[scala-library-2.12.18.jar:?]
	at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:172) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at com.marklogic.flux.impl.importdata.ImportJdbcCommand.loadDataset(ImportJdbcCommand.java:48) ~[flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.impl.AbstractCommand.readDataset(AbstractCommand.java:131) ~[flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.impl.AbstractCommand.execute(AbstractCommand.java:66) ~[flux-cli-1.4.0.jar:?]
	... 5 more

I dropped a copy of mljdbc jar in the ext folder, and used the following jdbc settings:

--jdbc-url
jdbc:marklogic://xxx:8145/?ssl=true&sslfactory=org.postgresql.ssl.jdbc4.LibPQFactory&sslmode=verify-ca&sslrootcert=/opt/nifi_auth/ca-bundle.crt
--jdbc-driver
com.marklogic.Driver

It is reaching MarkLogic for sure, but not understanding it. Also unsure if it is using the MarkLogic driver or not. Any suggestions?

[rjrudin]

Can you attach the driver jar that you've added to the ./ext folder in Flux? I can try a test with it.

The error message - "The length of varchar type cannot be negative" - suggests that when the driver asks for a table schema (presumably using column-info for an Optic query?), the server is returned a string/varchar column with a negative length.

[grtjn]

You can grab it here: https://github.com/marklogic-community/ml-jdbc-driver/blob/master/jar/mljdbc-42.1.4.jar

[rjrudin]

That community project is unsupported and hasn't been updated in several years. I would not expect it to work. You may have better luck with the regular Postgres driver, but that is not supported either.

[grtjn]

Yeah, would be nice if a more official jar would be released for MarkLogic, but it runs like a charm with all our other tools, even against ML 11.3.2.

[grtjn]

If it is possible to duplicate the Exporting Rows functionality as an Importing Rows feature, that would probably work much better :)

[grtjn]

postgress 42.7.7 jar (latest I could find) gives the exact same error. would be great if there is a way to get this to work, via jdbc or otherwise.

[grtjn]

I ran a test with export-json-lines-files with following parameters:

--query
op.fromView\(\'${Schema}\',\'${View}\'\)
--partitions
1
--batch-size
0

where fromView is pointing to a QBV. It returns an exception however:

Command failed; consider running the command with the --stacktrace option to see more error information. Error: Unable to run Optic query op.fromView('schema','view'); cause: Local message: failed to apply resource at internal/viewinfo: Internal Server Error. Server Message: SQL-NOCOLUMN: Column not found: rowID . See the MarkLogic server error log for further detail.

which makes sense, as QBVs themselves don't have a rowID column. Only TDEs have those. The QBVs do rely on TDEs underneath, but with faily complex logic to join, harmonize, and rename fields.

So I ran another test with fromSQL:

--query
op.fromSQL\(\'SELECT\ \*\ FROM\ ${Schema}.${View}\'\)

But that returns another NullPointerException:

2025-11-27 13:28:02 com.marklogic.flux: Will connect to MarkLogic host: xxx
2025-11-27 13:28:05 com.marklogic.flux: Using path: /data/test/Schema.View
2025-11-27 13:28:05 com.marklogic.spark: Optic query does not contain 'op.fromView', so will read rows in a single call to MarkLogic.
2025-11-27 13:28:05 com.marklogic.spark: Partition count: 1; number of requests that will be made to MarkLogic: 1
2025-11-27 13:28:10 com.marklogic.flux: Displaying stacktrace due to use of --stacktrace option
com.marklogic.flux.api.FluxException: null
	at com.marklogic.flux.impl.AbstractCommand.handleException(AbstractCommand.java:161) ~[flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.impl.AbstractCommand.execute(AbstractCommand.java:82) ~[flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.cli.Main.executeCommand(Main.java:138) ~[flux-cli-1.4.0.jar:?]
	at picocli.CommandLine.execute(CommandLine.java:2174) [picocli-4.7.6.jar:4.7.6]
	at com.marklogic.flux.cli.Main.run(Main.java:114) [flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.cli.Main.run(Main.java:79) [flux-cli-1.4.0.jar:?]
	at com.marklogic.flux.cli.Main.main(Main.java:75) [flux-cli-1.4.0.jar:?]
Caused by: java.lang.NullPointerException
	at org.apache.spark.sql.catalyst.expressions.codegen.UnsafeWriter.write(UnsafeWriter.java:110) ~[spark-catalyst_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) ~[?:?]
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43) ~[spark-sql_2.12-3.5.6.jar:3.5.6]
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.18.jar:?]
	at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.scheduler.Task.run(Task.scala:141) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:621) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64) ~[spark-common-utils_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61) ~[spark-common-utils_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:624) ~[spark-core_2.12-3.5.6.jar:3.5.6]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
	at java.base/java.lang.Thread.run(Thread.java:829) ~[?:?]

[rjrudin]

That sounds like it may be an error specific to a particular column. I did a quick test of using --query "op.fromSQL('SELECT * FROM Medical.Authors')", and that ran fine.

Try modifying the ./conf/log4j2.properties file and changing com.marklogic.spark to DEBUG. You'll then see a line of logging starting with something like this:

16:34:02.546 [Test worker] DEBUG com.marklogic.spark - Inferred schema from Optic columnInfo: StructType(StructField(Medical.Authors.CitationID,LongType,false)

That will should the schema based on the columnInfo result for your QBV. Perhaps there's a null value in a column that's marked as non-nullable?

To debug, I recommend slimming down the set of column returned by your QBV to help identify which column(s) is causing the issue.

[grtjn]

That gets me further. I am a bit surprised that a QBV can have non-nullable columns, as you cannot define columns.

More interestingly, we have a QBV that hard-coded puts some value to NULL (empty-sequence in practice), and it is exactly those columns that are returned as nullable false by op:column-info (which I can just get directly from QC). How on earth do I tell a QBV that those columns are nullable!??

FYI, fromSQL runs okay when I just select the first column, which is never null, so one step further again.