We had a pretty interesting experience trying to get performance and correctness by sending all of our messages between nodes using ucx-py and dask to send messages. The single threaded nature of python, the fact that dask is using torando.ioloop and we were seeing things like coroutines run at the same time if we were awaiting a ucx.send. It has been really hard to troubleshoot and the performance isn't there for us.

We need to send and receive messages in the c++ layer to remove the issues we had. Seeing as how we have often been hasty in trying to implement ucx as fast as possible we are going to try and be smart and slow the heck down. Hell if it takes 3 times as long to develop and 1/2 as long to debug we will come out ahead :).

I kind of envision a few classes like this

template < typename SerializerFunction, typename BufferCommunicator> 
class SenderClass{
   SenderClass(serializer, bufferCommunicator)

    //stuff like broadcast, send_to_node, 
}

template < typename DeserializerFunction, typename BufferAssembler >
class ReceiverClass{
    ReceiverClass(deserializer,bufferAssembler)
}

SerializerFunc ==> f(vector<column_views>) returns a list of views to rmm buffers, metadata for reassembling these buffers into views
DeserializerFunc ==> takes a list of buffers and metadata and gives us a unique_ptr to a BlazingTable
BufferAssembler ==> collects all the buffers of a message and associate them with their metadata
BufferCommunicator ==> can send a single buffer from one node to another and have it arrive at the other nodes buffer assembler

We can use a combination of these things to send a message and receive it on the other end with a listener that basses the buffer to the bufferAssembler, when the buffer assembler is done the deserializer converts it to a cudf::table and a Metadata that we can use to add the message to the appropriate class

blazingsql cannot process csv files larger than 3 GB, and the message "out of memory" is displayed. The number of GPUs is 4 and the total memory is 6 GB. The failure occurs whether a single GPU or multiple GPUs are used.

If multiple GPUs are used, the following information is displayed: distributed.nanny - WARNING - Restarting worker

Enabling this new env flag BLAZINGSQL_E2E_TEST_WITH_NULLS, I got:

TOTAL SUMMARY for test suite: 
PASSED: 1839/1945
FAILED: 106/1945
CRASH: 0/1945
TOTAL: 1945
saveLog = false
MAX DELTA: 192.0

Right now when you do: bc.sql() execution on the python script halts until that function call returns when the function returns with the result of the query. You used to be able to use the option return_futures but that feature is now obsolete due to https://github.com/BlazingDB/blazingsql/pull/1289

On the other hand https://github.com/BlazingDB/blazingsql/pull/1289 makes it easy to implement multiple concurrent queries.

This feature request is to propose an API and user experience for multiple concurrent queries support from a single BlazingContext.

The proposed API would be something as follows: Proposed API A

query0 = 'SELECT * FROM my_table where columnA > 0'
query1 = 'SELECT * FROM my_table where columnB < 0'
token0 = bc.sql(query0, return_token=True)  
token1 = bc.sql(query1, return_token=True)
result0 = bc.fetch(token0)
result1 = bc.fetch(token1)

In this case token0 and token1 would be int32s which are actually just the queryId. In this case bc.fetch would halt execution until the results are available. We would also implement a function (which would be optional) that would look like this: done = bc.is_query_done(token0) which would return a boolean, simply indicating if the query is done.

Other ways we could do this are: Proposed API B:

token0 = bc.async_sql(query)
is_done = bc.async_sql(token0, get_status=True)  #this is the optional is_query_done API
result = bc.async_sql(token0)  # here its the same API, but since we are passing in an int instead of a string we would know that we are getting the result

Proposed API C:

token0 = bc.sql(query, return_token=True)
is_done = bc.sql(token0, get_status=True)  #this is the optional is_query_done API
result = bc.sql(token0)  # here its the same API, but since we are passing in an int instead of a string we would know that we are getting the result

Feel free to propose other APIs.

This internally, this would just use the APIs that are now part of https://github.com/BlazingDB/blazingsql/pull/1289, which allow us to start a query, check its status and get the results. Internally what happens is that when multiple queries are running at the same time is that each query has its own graph, and each graph is generating compute tasks. The compute tasks are then processed by the executor as resources allow. Right now the tasks would be processed FIFO (with a certain amount of parallelism depending on resources and configuration). Eventually we can set prioritization policies for which tasks get done first. For example tasks from the first query to start are given priority, or tasks which are most likely to reduce memory pressure are prioritized, etc...

This PR adds CC, CXX, and CUDAHOSTCXX entries to the build.script_env section of the conda recipe, so that those environment variables get passed through the build.sh script and ultimately to CMake. This enables CMake to use the correct versions of gcc and g++ when compiling.

Additionally, it includes some fixes for the upstream cudf changes in https://github.com/rapidsai/cudf/pull/8142

Proposal 1 Direct approach using only create table semantic:

bc.create_table("dept_emp", "mysql://lucho:admin@localhost:3306/sampledb/dept_emp")
bc.create_table("titles", "postgres://luis:12345@localhost:3306/testdb/subchema/titles")

Proposal 2 2 step approach, similar to what we have with storage registration:

bc.mysql("mysqldb1", "mysql://lucho:admin@localhost:3306/sampledb")
bc.create_table("dept_emp", "mysqldb1/dept_emp")

bc.postgres("pgdb1", "postgres://luis:12345@localhost:3306/testdb")
bc.create_table("titles", "pgdb1/subchema/titles")

cc @williamBlazing @felipeblazing @rommelDB

Describe the bug Crash when using example from: https://blog.blazingdb.com/data-visualization-with-blazingsql-12095862eb73

Steps/Code to reproduce bug run sample code [(attached)]([url](url s3-test.py.txt ))

Expected behavior No illegal memory access exception.

Environment overview (please complete the following information)

• Environment location: Bare metal conda Python 3.7.7 (default, Mar 23 2020, 22:36:06) [GCC 7.3.0] :: Anaconda, Inc. on linux Debian 10 CUDA 10.2

• Method of cuDF install:

conda install -c blazingsql/label/cuda10.2 -c blazingsql -c rapidsai -c nvidia -c conda-forge -c defaults blazingsql python=3.7

Environment details PATH=/opt/miniconda3/bin:/opt/miniconda3/condabin:/usr/local/bin:/usr/bin:/bin:/usr/local/games:/usr/games:/usr/local/go/bin

Additional context Code attached. Other tests using blazingsql worked fine on this box.

Output:

listening: tcp://*:22758 2020-06-14T15:56:41Z|-78920688|TRACE|deregisterFileSystem: filesystem authority not found CacheDataLocalFile: /tmp/.blazing-temp-D63WqK6ZgzRBOMd0kxS4CzTDNC69hqAn1vlzzPGIjU8ijs78nLFqpShVKo8Qkdmm.orc terminate called after throwing an instance of 'thrust::system::system_error' what(): parallel_for failed: cudaErrorIllegalAddress: an illegal memory access was encountered distributed.nanny - WARNING - Restarting worker BlazingContext ready distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing

After crash, nvidia-smi shows below, main python process is hung:

+-----------------------------------------------------------------------------+
| NVIDIA-SMI 440.82       Driver Version: 440.82       CUDA Version: 10.2     |
|-------------------------------+----------------------+----------------------+
| GPU  Name        Persistence-M| Bus-Id        Disp.A | Volatile Uncorr. ECC |
| Fan  Temp  Perf  Pwr:Usage/Cap|         Memory-Usage | GPU-Util  Compute M. |
|===============================+======================+======================|
|   0  TITAN V             On   | 00000000:01:00.0 Off |                  N/A |
| 29%   43C    P8    26W / 250W |    640MiB / 12066MiB |      0%      Default |
+-------------------------------+----------------------+----------------------+
                                                                               
+-----------------------------------------------------------------------------+
| Processes:                                                       GPU Memory |
|  GPU       PID   Type   Process name                             Usage      |
|=============================================================================|
|    0     31262      C   python                                       627MiB |
+-----------------------------------------------------------------------------+

First time I ran it, it created a number of .orc files in /tmp before crashing with above error. Another time it gave:

listening: tcp://*:22170 BlazingContext ready 2020-06-14T16:59:14Z|-682139984|TRACE|deregisterFileSystem: filesystem authority not found distributed.nanny - WARNING - Restarting worker distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing Unable to start CUDA Context Traceback (most recent call last): File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/dask_cuda/initialize.py", line 108, in dask_setup numba.cuda.current_context() File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/devices.py", line 212, in get_context return _runtime.get_or_create_context(devnum) File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/devices.py", line 138, in get_or_create_context return self._get_or_create_context_uncached(devnum) File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/devices.py", line 153, in _get_or_create_context_uncached return self._activate_context_for(0) File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/devices.py", line 169, in _activate_context_for newctx = gpu.get_primary_context() File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/driver.py", line 529, in get_primary_context driver.cuDevicePrimaryCtxRetain(byref(hctx), self.id) File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/driver.py", line 295, in safe_cuda_api_call self._check_error(fname, retcode) File "/opt/miniconda3/envs/py37/lib/python3.7/site-packages/numba/cuda/cudadrv/driver.py", line 330, in _check_error raise CudaAPIError(retcode, msg) numba.cuda.cudadrv.driver.CudaAPIError: [304] Call to cuDevicePrimaryCtxRetain results in CUDA_ERROR_OPERATING_SYSTEM pure virtual method called terminate called without an active exception

Edit: Tried accessing the files locally instead of from s3 and reproduced same error. As soon as memory fills up or after several orc files created (varies) gets illegal memory access or worker dies. If the worker restarts successfully it does not appear to process anything. Using non-dask/cluster version of BlazingContext it says 'Killed' as soon as it runs out of memory on the GPU. Processing only one input file works fine as it does not run out of memory.

(rapids_blazing) sh-4.2$ conda list | grep blazing blazingdb-toolchain 0.4.0 py37hf484d3e_0 blazingsql blazingsql-calcite 0.4.0 py37_0 blazingsql blazingsql-communication 0.4.0 py37_80 blazingsql blazingsql-io 0.4.0 py37_31 blazingsql blazingsql-orchestrator 0.4.0 py37_19 blazingsql blazingsql-protocol 0.4.0 py37_25 blazingsql blazingsql-python 0.4.0 cuda10.0_py37_14 blazingsql/label/cuda10.0 blazingsql-ral 0.4.0 cuda10.0_py37_5 blazingsql/label/cuda10.0

Yesterday I was able to create a table based on a cudf but today I'm having some errors. I have already created again the whole instance and did again all the steps but I'm facing some errors similar to the following: b'In function ddlCreateTableService: cannot create the table: Could not create table'

There is no more information. Any ideas how I can trace the error? In addition, it seems in version 0.4.2 there will be some changes in how BlazingContext launches processes, could it be related to this? When this release will be conda installable?

If I execute again BlazingContext() and try to create the table I can get two kinds of different errors: 1) Already connected to the Orchestrator b'In function ddlCreateTableService: cannot create the table: Connection to server failed.'

WARNING: blazingsql-orchestrator was not automativally started, its probably already running WARNING: blazingsql-engine was not automativally started, its probably already running WARNING: blazingsql-algebra was not automativally started, its probably already running Already connected to the Orchestrator Unexpected error on create_table, can only concatenate str (not "tuple") to str

This PR:

• Implements the REPLACE operator for string columns using cudf::strings::replace
• Adds a new end-to-end test in stringsTests.py, and updates the runTest.py
• Refactors the removal of the string encaspulation characters (i.e., the single quotes in LIKE '%the%') in several parts of LogicalProjection.cpp to use a string utility function

If the implementation looks fine, I'll push the new parquet file to https://github.com/BlazingDB/blazingsql-testing-files and update the CHANGELOG to unblock gpuCI.

This closes https://github.com/BlazingDB/blazingsql/issues/1175

• [x] passes e2e tests locally
• [x] PR to update testing files https://github.com/BlazingDB/blazingsql-testing-files/pull/1

from pyspark.sql import SparkSession
from blazingsql import BlazingContext
import pandas as pd
​
​
# spark = SparkSession.builder \
#     .master("local") \
#     .getOrCreate()
​
# bc = BlazingContext()
​
​
df = pd.DataFrame({
    "a": ["Felipe", "William", "Rodrigo"],
    "b": [2, 4, 6],
    "c": ["2020-11-20", "2020-11-19", "2020-11-18"]
})
​
bc.create_table("df", df)
sdf = spark.createDataFrame(df)
sdf.createOrReplaceTempView("df")
​
query = """
SELECT
    a,
    REPLACE(a, 'i', '##') as a_new,
    c,
    REPLACE(c, '2020', '1999') as c_new
FROM df
"""
​
spark.sql(query).show()
​
print(bc.explain(query))
print(bc.sql(query))
+-------+---------+----------+----------+
|      a|    a_new|         c|     c_new|
+-------+---------+----------+----------+
| Felipe|  Fel##pe|2020-11-20|1999-11-20|
|William|W##ll##am|2020-11-19|1999-11-19|
|Rodrigo| Rodr##go|2020-11-18|1999-11-18|
+-------+---------+----------+----------+

LogicalProject(a=[$0], a_new=[REPLACE($0, 'i', '##')], c=[$1], c_new=[REPLACE($1, '2020', '1999')])
  BindableTableScan(table=[[main, df]], projects=[[0, 2]], aliases=[[a, a_new, c, c_new]])

         a      a_new           c       c_new
0   Felipe    Fel##pe  2020-11-20  1999-11-20
1  William  W##ll##am  2020-11-19  1999-11-19
2  Rodrigo   Rodr##go  2020-11-18  1999-11-18

Right now Kernels handle distribution and ensuring completeness so that they continue when they have to communicate. Here is an example of what that looks like in aggregation.

Below we are iterating through batches that this kernel gets from its input cache, partitioning them and sending each node its corresponding partition. We store a count of how many partitions we sent to each node and how many we kept for ourselves.

while (input.wait_for_next()) {
    auto batch = input.next();
    CudfTableView batch_view = batch->view();
    std::vector<CudfTableView> partitioned;
    std::unique_ptr<CudfTable> hashed_data; // Keep table alive in this scope
    if (batch_view.num_rows() > 0) {
        std::vector<cudf::size_type> hased_data_offsets;
        std::tie(hashed_data, hased_data_offsets) = cudf::hash_partition(batch->view(), columns_to_hash, num_partitions);
        // the offsets returned by hash_partition will always start at 0, which is a value we want to ignore for cudf::split
        std::vector<cudf::size_type> split_indexes(hased_data_offsets.begin() + 1, hased_data_offsets.end());
        partitioned = cudf::split(hashed_data->view(), split_indexes);
    } else {
        //  copy empty view
        for (auto i = 0; i < num_partitions; i++) {
            partitioned.push_back(batch_view);
        }
    }

    ral::cache::MetadataDictionary metadata;
    for(int i = 0; i < this->context->getTotalNodes(); i++ ){
        auto partition = std::make_unique<ral::frame::BlazingTable>(partitioned[i], batch->names());
        if (this->context->getNode(i) == self_node){
            this->output_.get_cache()->addToCache(std::move(partition),"",true);
            node_count[self_node.id()]++;
        } else {
            node_count[this->context->getNode(i).id()]++;
            output_cache->addCacheData(std::make_unique<ral::cache::GPUCacheDataMetaData>(std::move(partition), metadata),"",true);
        }
    }
    }
    batch_count++;
}

After this code executes we send each node a count of how many partitions we sent them.

auto self_node = ral::communication::CommunicationData::getInstance().getSelfNode();
auto nodes = context->getAllNodes();
std::string worker_ids = "";


for(std::size_t i = 0; i < nodes.size(); ++i) {
    if(!(nodes[i] == self_node)) {
        ral::cache::MetadataDictionary metadata;
        messages_to_wait_for.push_back(metadata.get_values()[ral::cache::QUERY_ID_METADATA_LABEL] + "_" +
                                metadata.get_values()[ral::cache::KERNEL_ID_METADATA_LABEL] +	"_" +
                                metadata.get_values()[ral::cache::WORKER_IDS_METADATA_LABEL]);
        this->query_graph->get_output_cache()->addCacheData(
            std::unique_ptr<ral::cache::GPUCacheData>(new 
               ral::cache::GPUCacheDataMetaData(ral::utilities::create_empty_table({}, {}), metadata)),"",true);
    }
}

Then we collect all of the partition counts from each worker node. After this we sum them up and wait for our output cache to have that many partitions before we can say this kernel is finished.

auto self_node = ral::communication::CommunicationData::getInstance().getSelfNode();
int total_count = node_count[self_node.id()];
for (auto message : messages_to_wait_for){
    auto meta_message = this->query_graph->get_input_cache()->pullCacheData(message);
    total_count += std::stoi(static_cast<ral::cache::GPUCacheDataMetaData *>(meta_message.get())->getMetadata().get_values()[ral::cache::PARTITION_COUNT]);
}
this->output_cache()->wait_for_count(total_count);

We want to abstract away a few of the things that are happening here. We are often following this pattern of spreading data out and then theres a barrier to be able to continue. We want to remove this code from the kernel run function itself and have a more generic way of saying things like

As we discuss and implement the movement towards scheduling tasks to be run we need to have primitives that can do things like :

• create a broadcast to all and expect broadcast from all primitive
• create a method for preventing tasks from either being scheduled or run by the scheduler until some kind of condition is met (e.g. wait_for_count but disassociated from the actual run function so it is something that can be "injected" preferably through something like composition).

This PR contains fixes for issues of building, cc @romulo-auccapuclla This PR also contains fixes for new arrow API 4.0.1 (due to cudf-nightly) This PR contains a fix due to https://github.com/rapidsai/cudf/pull/8692 (related to the conda env name). Note: Something that worries me is that the HiveFileTest lately crashes randomly with a std::bad_alloc(query 01 CSV, that was commented out).

Bumps calcite-core from 1.23.0 to 1.32.0.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

Describe the bug Cannot import BlazingContext when processor type unknown.

Steps/Code to reproduce bug Code and output from ipython (personal info hidden).

In [1]: from blazingsql import BlazingContext
---------------------------------------------------------------------------
FileNotFoundError                         Traceback (most recent call last)
<ipython-input-7-0b19b5b41f48> in <module>
----> 1 from blazingsql import BlazingContext

~/miniconda3/envs/blazingsql/lib/python3.7/site-packages/blazingsql/__init__.py in <module>
      1 from pyblazing.apiv2 import S3EncryptionType
      2 from pyblazing.apiv2 import DataType
----> 3 from pyblazing.apiv2.context import BlazingContext
      4
      5 from cio import getProductDetailsCaller

~/miniconda3/envs/blazingsql/lib/python3.7/site-packages/pyblazing/apiv2/context.py in <module>
    105         )
    106
--> 107 jpype.startJVM("-ea", convertStrings=False, jvmpath=jvm_path)
    108 # jpype.startJVM()
    109

~/miniconda3/envs/blazingsql/lib/python3.7/site-packages/jpype/_core.py in startJVM(*args, **kwargs)
    225     try:
    226         _jpype.startup(jvmpath, tuple(args),
--> 227                        ignoreUnrecognized, convertStrings, interrupt)
    228         initializeResources()
    229     except RuntimeError as ex:

FileNotFoundError: [Errno 2] JVM DLL not found: /home/{my_username}/miniconda3/envs/blazingsql/lib/server/libjvm.so


In [2]: !uname -p
unknown

Expected behavior Should be imported without any errors.

Environment overview (please complete the following information)

• Environment location: Bare-metal
• Method of BlazingSQL install: conda
• BlazingSQL Version

BlazingSQL version (git hash): 13618d177a37bd34bb20ac832fb8a14f8243ff5c
BlazingSQL branch name: HEAD
BlazingSQL branch tag: v21.08.02
BlazingSQL build id: 0
BlazingSQL compiler version: GNU /usr/local/gcc9/bin/g++ 9.4.0
BlazingSQL cuda flags: -Xcompiler -Wno-parentheses -gencode=arch=compute_60,code=sm_60 -gencode=arch=compute_70,code=sm_70 -gencode=arch=compute_75,code=sm_75 -gencode=arch=compute_75,code=compute_75 --expt-extended-lambda --expt-relaxed-constexpr -Werror=cross-execution-space-call -Xcompiler -Wall,-Wno-error=deprecated-declarations --default-stream=per-thread -DHT_DEFAULT_ALLOCATOR
BlazingSQL Operating system kernel: Linux-5.4.0-1054-aws
BlazingSQL Operating system architecture: x86_64
BlazingSQL Linux Operating system release: NAME=CentOS Linux|VERSION=7 (Core)|ID=centos|ID_LIKE=rhel fedora|VERSION_ID=7|PRETTY_NAME=CentOS Linux 7 (Core)|ANSI_COLOR=031|CPE_NAME=cpe:/o:centos:centos:7|HOME_URL=[https://www.centos.org/|BUG_REPORT_URL=https://bugs.centos.org/||CENTOS_MANTISBT_PROJECT=CentOS-7|CENTOS_MANTISBT_PROJECT_VERSION=7|REDHAT_SUPPORT_PRODUCT=centos|REDHAT_SUPPORT_PRODUCT_VERSION=7|](https://www.centos.org/%7CBUG_REPORT_URL=https://bugs.centos.org/%7C%7CCENTOS_MANTISBT_PROJECT=CentOS-7%7CCENTOS_MANTISBT_PROJECT_VERSION=7%7CREDHAT_SUPPORT_PRODUCT=centos%7CREDHAT_SUPPORT_PRODUCT_VERSION=7%7C)

----For BlazingSQL Developers---- Suspected source of the issue https://github.com/BlazingDB/blazingsql/blob/branch-21.08/pyblazing/pyblazing/apiv2/context.py#L70

machine_processor = platform.processor()

if machine_processor in ("x86_64", "x64"):
    machine_processor = "amd64"

when the uname -p is unknown, platform.processor() equals to '', thus machine_processor is empty, which leads to wrong jvm lib path.

Describe the bug Not able to open any app on the app deployment

https://app.blazingsql.com/jupyter/user-redirect/lab/workspaces/auto-b/tree/Welcome_to_BlazingSQL_Notebooks/welcome.ipynb

Steps/Code to reproduce bug Go to browser and try to open https://app.blazingsql.com Error: Site could not be reached

Expected behavior Should be able to open the website

Other design considerations It is very difficult to set up blazing SQL on google collab. It is giving me too many version compatibility issues. If someone can help me with it that would work as well

Bumps liquibase-core from 3.6.2 to 4.8.0.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

When I am trying to run blazingsql in google colab I am finding the following error Link of colab Notebook: https://blog.blazingdb.com/blazingsql-rapids-ai-now-free-on-google-colab-b8646f1ea948 <module 'subprocess' from '/usr/lib/python3.7/subprocess.py'>

FileNotFoundError Traceback (most recent call last) in () 8 import subprocess 9 print(subprocess) ---> 10 subprocess.Popen(['blazingsql-orchestrator', '9100', '8889', '127.0.0.1', '8890'],stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) 11 subprocess.Popen(['java', '-jar', '/usr/local/lib/blazingsql-algebra.jar', '-p', '8890']) 12 import pyblazing.apiv2.context as cont

1 frames /usr/lib/python3.7/subprocess.py in _execute_child(self, args, executable, preexec_fn, close_fds, pass_fds, cwd, env, startupinfo, creationflags, shell, p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite, restore_signals, start_new_session) 1549 if errno_num == errno.ENOENT: 1550 err_msg += ': ' + repr(err_filename) -> 1551 raise child_exception_type(errno_num, err_msg, err_filename) 1552 raise child_exception_type(err_msg) 1553

FileNotFoundError: [Errno 2] No such file or directory: 'blazingsql-orchestrator': 'blazingsql-orchestrator'